from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime, time
from typing import Any, Dict, List, Literal, Optional, TypedDict
from ergani.typings import (
LateDeclarationJustificationType,
OvertimeJustificationType,
ScheduleWorkType,
WorkCardMovementType,
)
from ergani.utils import (
format_date,
format_datetime,
format_time,
get_day_of_week,
get_ergani_late_declaration_justification,
get_ergani_overtime_cancellation,
get_ergani_overtime_justification,
get_ergani_work_type,
get_ergani_workcard_movement_type,
)
[docs]
@dataclass(frozen=True)
class CurrentWorkforceRequest:
afm: str | None = None
def serialize(self) -> Dict[str, Any]:
payload: Dict[str, Any] = {}
if self.afm is not None:
payload["afm"] = self.afm
return payload
[docs]
@dataclass
class CurrentWorkforceRecord:
employee_tax_identification_number: str | None = None
employee_last_name: str | None = None
employee_first_name: str | None = None
employee_father_first_name: str | None = None
employee_mother_first_name: str | None = None
birth_date: str | None = None
sex: str | None = None
nationality: str | None = None
marital_status: str | None = None
number_of_children: int | None = None
tax_office: str | None = None
unemployment_card_code: str | None = None
social_security_registry_number: str | None = None
social_security_number: str | None = None
address: str | None = None
postal_code: str | None = None
phone_number: str | None = None
kallikratis_municipal_code: str | None = None
underage_work_book_number: str | None = None
identity_document_type: str | None = None
identity_document_number: str | None = None
identity_document_issuing_authority: str | None = None
identity_document_issue_date: str | None = None
residence_permit_installment: str | None = None
residence_permit_installment_number: str | None = None
residence_permit_approval: str | None = None
residence_permit_approval_number: str | None = None
residence_permit_visa: str | None = None
residence_permit_visa_number: str | None = None
branch_number: int | None = None
employment_start_date: str | None = None
specialty: str | None = None
employee_classification: str | None = None
profession_code: str | None = None
weekly_workdays: str | None = None
prior_experience: str | None = None
employment_relationship: str | None = None
responsible_position: str | None = None
employment_status: str | None = None
weekly_hours: str | None = None
working_schedule: str | None = None
break_schedule: str | None = None
workplace: str | None = None
workplace_comments: str | None = None
wage_payment_frequency: str | None = None
unpredictable_work_schedule: str | None = None
on_demand_days_and_hours: str | None = None
on_demand_minimum_notification: str | None = None
on_demand_notes: str | None = None
mandatory_training: str | None = None
applicable_collective_agreement: str | None = None
applicable_collective_agreement_comments: str | None = None
working_time_arrangement: str | None = None
working_time_arrangement_comments: str | None = None
gross_pay: str | None = None
hourly_pay: str | None = None
primary_insurance: str | None = None
supplementary_insurance: str | None = None
additional_insurance_benefits: str | None = None
trial_period: str | None = None
borrowing_company_tax_identification_number: str | None = None
change_date: str | None = None
education_level: str | None = None
professional_education: str | None = None
pc_provided: str | None = None
working_time_digital_organization: str | None = None
full_employment_hours: str | None = None
break_minutes: int | None = None
break_within_schedule: str | None = None
working_card: str | None = None
flexible_working_hours: str | None = None
last_modified_date: str | None = None
raw_payload: Dict[str, Any] = field(default_factory=dict)
@classmethod
def parse_many(cls, payload: Any) -> List[CurrentWorkforceRecord]:
workforce_payloads = cls._parse_payloads(payload)
return [cls.parse(payload) for payload in workforce_payloads]
@classmethod
def parse(cls, payload: Any) -> CurrentWorkforceRecord:
if not isinstance(payload, dict):
raise ValueError(
"Expected current workforce record payload to be an object"
)
return cls(
employee_tax_identification_number=payload.get("afm"),
employee_last_name=payload.get("Eponimo"),
employee_first_name=payload.get("Onoma"),
employee_father_first_name=payload.get("OnomaPatera"),
employee_mother_first_name=payload.get("OnomaMiteras"),
birth_date=payload.get("BirthDate"),
sex=payload.get("Sex"),
nationality=payload.get("Nationality"),
marital_status=payload.get("MaritalStatus"),
number_of_children=_parse_int(payload.get("NumChildren")),
tax_office=payload.get("Doy"),
unemployment_card_code=payload.get("CodeAnergias"),
social_security_registry_number=payload.get("AmIka"),
social_security_number=payload.get("Amka"),
address=payload.get("Dieythinsi"),
postal_code=payload.get("Tk"),
phone_number=payload.get("Tilefwno"),
kallikratis_municipal_code=payload.get("Kallikratis"),
underage_work_book_number=payload.get("ArVivliouAnilikou"),
identity_document_type=payload.get("TyposTaytotitas"),
identity_document_number=payload.get("ArTaytotitas"),
identity_document_issuing_authority=payload.get("EkdousaArxi"),
identity_document_issue_date=payload.get("DateEkdosis"),
residence_permit_installment=payload.get("ResPermitInst"),
residence_permit_installment_number=payload.get("ResPermitInstAr"),
residence_permit_approval=payload.get("ResPermitAp"),
residence_permit_approval_number=payload.get("ResPermitApAr"),
residence_permit_visa=payload.get("ResPermitVisa"),
residence_permit_visa_number=payload.get("ResPermitVisaAr"),
branch_number=_parse_int(payload.get("PararthmaAa")),
employment_start_date=payload.get("DateFrom"),
specialty=payload.get("Eidikothta"),
employee_classification=payload.get("asXaraktirismos"),
profession_code=payload.get("Step"),
weekly_workdays=payload.get("WeekDays"),
prior_experience=payload.get("Proipiresia"),
employment_relationship=payload.get("SxesiApasxolisis"),
responsible_position=payload.get("ResponsiblePosition"),
employment_status=payload.get("KathestosApasxolisis"),
weekly_hours=payload.get("WeekHours"),
working_schedule=payload.get("Orario"),
break_schedule=payload.get("Dialeimma"),
workplace=payload.get("ToposErgasias"),
workplace_comments=payload.get("ToposErgasiasComments"),
wage_payment_frequency=payload.get("XronosKatabolisApodoxwn"),
unpredictable_work_schedule=payload.get("MhProblepsimoProgrammaErgasias"),
on_demand_days_and_hours=payload.get("ParaggeliaHmeresHours"),
on_demand_minimum_notification=payload.get("ParaggeliaMinNotification"),
on_demand_notes=payload.get("ParaggeliaNotes"),
mandatory_training=payload.get("IpoxreotikiKatartisi"),
applicable_collective_agreement=payload.get("EfarmosteaSyllogikiSymbasi"),
applicable_collective_agreement_comments=payload.get(
"EfarmosteaSyllogikiSymbasiComments"
),
working_time_arrangement=payload.get("Dieythetisi"),
working_time_arrangement_comments=payload.get("DieythetisiComments"),
gross_pay=payload.get("Apodoxes"),
hourly_pay=payload.get("HourApodoxes"),
primary_insurance=payload.get("KyriaAsfalisi"),
supplementary_insurance=payload.get("EpikourikiAsfalisi"),
additional_insurance_benefits=payload.get("ProsthetesAsfalistikesParoxes"),
trial_period=payload.get("TrialPeriod"),
borrowing_company_tax_identification_number=payload.get("BorrowCompanyAfm"),
change_date=payload.get("DateMetabolhs"),
education_level=payload.get("EpipedoMorfosis"),
professional_education=payload.get("ProfessionalEducation"),
pc_provided=payload.get("Pc"),
working_time_digital_organization=payload.get(
"WorkingTimeDigitalOrganization"
),
full_employment_hours=payload.get("FullEmploymentHours"),
break_minutes=_parse_int(payload.get("DialeimmaMinutes")),
break_within_schedule=payload.get("DialeimmaEntosWrariou"),
working_card=payload.get("WorkingCard"),
flexible_working_hours=payload.get("EueliktoWrario"),
last_modified_date=payload.get("LastModifiedDate"),
raw_payload=dict(payload),
)
@classmethod
def _parse_payloads(cls, payload: Any) -> List[Dict[str, Any]]:
if payload is None:
return []
if isinstance(payload, dict) and "EX_BASE_05" in payload:
payload = payload["EX_BASE_05"]
if isinstance(payload, dict) and "Cur" in payload:
payload = payload["Cur"]
if not isinstance(payload, list):
raise ValueError("Expected current workforce response payload to be a list")
return payload
def _parse_int(value: Any) -> int | None:
if value is None or value == "":
return None
try:
return int(value)
except (TypeError, ValueError):
return None
[docs]
@dataclass
class WorkCard:
"""
Represents a work card entry for an employee
Attributes:
employee_tax_identification_number (str): The employee's tax identification number
employee_last_name (str): The last name of the employee
employee_first_name (str): The first name of the employee
work_card_movement_type (WorkCardMovementType): The type of work card movement
work_card_submission_date (date): The date the work card was submitted
work_card_movement_datetime (datetime): The exact date and time of the work card movement
late_declaration_justification (Optional[LateDeclarationJustificationType]): The justification for the late declaration of the work card movement
"""
employee_tax_identification_number: str
employee_last_name: str
employee_first_name: str
work_card_movement_type: WorkCardMovementType
work_card_submission_date: date
work_card_movement_datetime: datetime
late_declaration_justification: Optional[LateDeclarationJustificationType] = None
def serialize(self):
return {
"f_afm": self.employee_tax_identification_number,
"f_eponymo": self.employee_last_name,
"f_onoma": self.employee_first_name,
"f_type": get_ergani_workcard_movement_type(self.work_card_movement_type),
"f_reference_date": self.work_card_submission_date.isoformat(),
"f_date": format_datetime(self.work_card_movement_datetime),
"f_aitiologia": get_ergani_late_declaration_justification(
self.late_declaration_justification
),
}
[docs]
@dataclass
class CompanyWorkCard:
"""
Represents work card entries that are issued on a single business branch
Attributes:
employer_tax_identification_number (str): The employer's tax identification number
business_branch_number (int): The number identifying the specific business branch
comments (Optional[str]): Additional comments related to the work cards
card_details (List[WorkCard]): A list of `WorkCard` entries for the business branch
"""
employer_tax_identification_number: str
business_branch_number: int
comments: Optional[str] = ""
card_details: List[WorkCard] = field(default_factory=list)
def serialize(self):
return {
"f_afm_ergodoti": self.employer_tax_identification_number,
"f_aa": self.business_branch_number,
"f_comments": self.comments,
"Details": {
"CardDetails": [
work_card.serialize() for work_card in self.card_details
]
},
}
[docs]
@dataclass
class Overtime:
"""
Represents an overtime entry for an employee
Attributes:
employee_tax_identification_number (str): The employee's tax identification number
employee_social_security_number (str): The employee's social security number
employee_last_name (str): The last name of the employee
employee_first_name (str): The first name of the employee
overtime_date (date): The date of the overtime
overtime_start_time (time): The start time of the overtime period
overtime_end_time (time): The end time of the overtime period
overtime_cancellation (bool): Indicates if the overtime was cancelled or not
employee_profession_code (str): The profession code of the employee
overtime_justification (OvertimeJustificationType): The justification for the overtime
weekly_workdays_number (Literal[5, 6]): The number of the employee's working days in a week
asee_approval (Optional[str]): The ASEE aproval
"""
employee_tax_identification_number: str
employee_social_security_number: str
employee_last_name: str
employee_first_name: str
overtime_date: date
overtime_start_time: time
overtime_end_time: time
overtime_cancellation: bool
employee_profession_code: str
overtime_justification: OvertimeJustificationType
weekly_workdays_number: Literal[5, 6]
asee_approval: Optional[str] = ""
def serialize(self):
return {
"f_afm": self.employee_tax_identification_number,
"f_amka": self.employee_social_security_number,
"f_eponymo": self.employee_last_name,
"f_onoma": self.employee_first_name,
"f_date": format_date(self.overtime_date),
"f_from": format_time(self.overtime_start_time),
"f_to": format_time(self.overtime_end_time),
"f_cancellation": get_ergani_overtime_cancellation(
self.overtime_cancellation
),
"f_step": self.employee_profession_code,
"f_reason": get_ergani_overtime_justification(self.overtime_justification),
"f_weekdates": self.weekly_workdays_number,
"f_asee": self.asee_approval,
}
[docs]
@dataclass
class CompanyOvertime:
"""
Represents overtime entries that are issued on a single business branch
Attributes:
business_branch_number (int): The number identifying the specific business branch
sepe_service_code (str): The SEPE service code
business_primary_activity_code (str): The primary activity code of the business
business_branch_activity_code (str): The activity code for the specific branch
kallikratis_municipal_code (str): The kallikratis municipal code
legal_representative_tax_identification_number (str): Tax identification number of the legal representative
employee_overtimes (List[Overtime]): A list of `Overtime` entries for employees
related_protocol_id (Optional[str]): Related protocol ID
related_protocol_date (Optional[date]): The date of the related protocol
employer_organization (Optional[str]): The employer's organization name
business_secondary_activity_code_1 (Optional[str]): Secondary activity code 1
business_secondary_activity_code_2 (Optional[str]): Secondary activity code 2
business_secondary_activity_code_3 (Optional[str]): Secondary activity code 3
business_secondary_activity_code_4 (Optional[str]): Secondary activity code 4
comments (Optional[str]): Additional comments related to the overtime entries
"""
business_branch_number: int
sepe_service_code: str
business_primary_activity_code: str
business_branch_activity_code: str
kallikratis_municipal_code: str
legal_representative_tax_identification_number: str
employee_overtimes: List[Overtime] = field(default_factory=list)
related_protocol_id: Optional[str] = ""
related_protocol_date: Optional[date] = None
employer_organization: Optional[str] = ""
business_secondary_activity_code_1: Optional[str] = ""
business_secondary_activity_code_2: Optional[str] = ""
business_secondary_activity_code_3: Optional[str] = ""
business_secondary_activity_code_4: Optional[str] = ""
comments: Optional[str] = ""
def serialize(self):
return {
"f_aa_pararthmatos": self.business_branch_number,
"f_rel_protocol": self.related_protocol_id,
"f_rel_date": format_date(self.related_protocol_date),
"f_ypiresia_sepe": self.sepe_service_code,
"f_ergodotikh_organwsh": self.employer_organization,
"f_kad_kyria": self.business_primary_activity_code,
"f_kad_deyt_1": self.business_secondary_activity_code_1,
"f_kad_deyt_2": self.business_secondary_activity_code_2,
"f_kad_deyt_3": self.business_secondary_activity_code_3,
"f_kad_deyt_4": self.business_secondary_activity_code_4,
"f_kad_pararthmatos": self.business_branch_activity_code,
"f_kallikratis_pararthmatos": self.kallikratis_municipal_code,
"f_comments": self.comments,
"f_afm_proswpoy": self.legal_representative_tax_identification_number,
"Ergazomenoi": {
"OvertimeErgazomenosDate": [
overtime.serialize() for overtime in self.employee_overtimes
]
},
}
[docs]
@dataclass
class WorkdayDetails:
"""
Represents details of an employee's workday
Attributes:
work_type (ScheduleWorkType): The type of an employee's work schedule
start_time (time): The start time of the workday
end_time (time): The end time of the workday
"""
work_type: ScheduleWorkType
start_time: time
end_time: time
def serialize(self):
return {
"f_type": get_ergani_work_type(self.work_type),
"f_from": format_time(self.start_time),
"f_to": format_time(self.end_time),
}
[docs]
@dataclass
class EmployeeDailySchedule:
"""
Represents a daily schedule entry for an employee
Attributes:
employee_tax_identification_number (str): The employee's tax identification number
employee_last_name (str): The employee's last name
employee_first_name (str): The employee's first name
schedule_date (date): The date of the schedule
workday_details (List[WorkdayDetails]): A list of workday detail entries for the employee
"""
employee_tax_identification_number: str
employee_last_name: str
employee_first_name: str
schedule_date: date
workday_details: List[WorkdayDetails] = field(default_factory=list)
def serialize(self):
return {
"f_afm": self.employee_tax_identification_number,
"f_eponymo": self.employee_last_name,
"f_onoma": self.employee_first_name,
"f_date": format_date(self.schedule_date),
"ErgazomenosAnalytics": {
"ErgazomenosWTOAnalytics": [
workday_detail.serialize()
for workday_detail in self.workday_details
]
},
}
[docs]
@dataclass
class CompanyDailySchedule:
"""
Represents daily schedule entries that are issued on a single business branch
Attributes:
business_branch_number (int): The number identifying the business branch
start_date (Optional[date]): The start date of the schedule
end_date (Optional[date]): The end date of the schedule period
employee_schedules (List[EmployeeDailySchedule]): A list of daily schedules for employees
related_protocol_id (Optional[str]): The ID of the related protocol
related_protocol_date (Optional[date]): The date of the related protocol
comments (Optional[str]): Additional comments regarding the daily schedule entries
"""
business_branch_number: int
start_date: Optional[date] = None
end_date: Optional[date] = None
employee_schedules: List[EmployeeDailySchedule] = field(default_factory=list)
related_protocol_id: Optional[str] = ""
related_protocol_date: Optional[date] = None
comments: Optional[str] = ""
def serialize(self):
return {
"f_aa_pararthmatos": self.business_branch_number,
"f_rel_protocol": self.related_protocol_id,
"f_rel_date": format_date(self.related_protocol_date),
"f_comments": self.comments,
"f_from_date": format_date(self.start_date),
"f_to_date": format_date(self.end_date),
"Ergazomenoi": {
"ErgazomenoiWTO": [
employee_schedule.serialize()
for employee_schedule in self.employee_schedules
]
},
}
[docs]
@dataclass
class EmployeeWeeklySchedule:
"""
Represents a weekly schedule entry for an employee
Attributes:
employee_tax_identification_number (str): The employee's tax identification number
employee_last_name (str): The employee's last name
employee_first_name (str): The employee's first name
schedule_date (date): The date of the schedule
workday_details (List[WorkdayDetails]): A list of workday detail entries for the week
"""
employee_tax_identification_number: str
employee_last_name: str
employee_first_name: str
schedule_date: date
workday_details: List[WorkdayDetails] = field(default_factory=list)
def serialize(self):
return {
"f_afm": self.employee_tax_identification_number,
"f_eponymo": self.employee_last_name,
"f_onoma": self.employee_first_name,
"f_day": get_day_of_week(self.schedule_date),
"ErgazomenosAnalytics": {
"ErgazomenosWTOAnalytics": [
workday_detail.serialize()
for workday_detail in self.workday_details
]
},
}
[docs]
@dataclass
class CompanyWeeklySchedule:
"""
Represents weekly schedule entries that are issued on a single business branch
Attributes:
business_branch_number (int): The number identifying the business branch
start_date (date): The start date of the weekly schedule
end_date (date): The end date of the weekly schedule
employee_schedules (List[EmployeeWeeklySchedule]): A list of weekly schedules for employees
related_protocol_id (Optional[str]): The ID of the related protocol
related_protocol_date (Optional[date]): The date of the related protocol
comments (Optional[str]): Additional comments regarding the weekly schedule entries
"""
business_branch_number: int
start_date: date
end_date: date
employee_schedules: List[EmployeeWeeklySchedule] = field(default_factory=list)
related_protocol_id: Optional[str] = ""
related_protocol_date: Optional[date] = None
comments: Optional[str] = ""
def serialize(self):
return {
"f_aa_pararthmatos": self.business_branch_number,
"f_rel_protocol": self.related_protocol_id,
"f_rel_date": format_date(self.related_protocol_date),
"f_comments": self.comments,
"f_from_date": format_date(self.start_date),
"f_to_date": format_date(self.end_date),
"Ergazomenoi": {
"ErgazomenoiWTO": [
employee_schedule.serialize()
for employee_schedule in self.employee_schedules
]
},
}
[docs]
class SubmissionResponse(TypedDict):
"""
Represents a submission response from the Ergani API
Attributes:
submission_id (str): The unique identifier of the submission
protocol (str): The protocol associated with the submission
submission_date (datetime): The datetime of the submission
"""
submission_id: str
protocol: str
submission_date: datetime