pull/29/head
ahmed.mujtaba 2026-08-27 16:20:12 +05:00
parent d5879a0617
commit fe80fa9fdc
3 changed files with 173 additions and 259 deletions

View File

@ -189,6 +189,81 @@ class FormData(SQLModel, table=True):
await session.commit() await session.commit()
return {"deleted": deleted, "inserted": inserted} return {"deleted": deleted, "inserted": inserted}
@staticmethod
def _cell(data: dict, key: str):
"""Sheet cell → stripped str, or None if missing/blank."""
value = data.get(key)
if value is None:
return None
text = str(value).strip()
return text if text else None
@classmethod
def from_sheet_row(cls, sheet: str, row_number: int, data: dict) -> dict:
"""Build FormData kwargs from one sheet row dict (exact header keys, no aliases).
Year of Graduation: prefer the second column when present; else the first;
else None. Duplicate headers are renamed Year of Graduation_1 by normalise_headers.
"""
from g_sheet.plugins import parse_date, parse_date_time, parse_salary
first_year = cls._cell(data, "Year of Graduation")
second_year = cls._cell(data, "Year of Graduation_1")
if second_year:
entry_year = second_year
elif first_year:
entry_year = first_year
else:
entry_year = None
timestamp_raw = data.get("Timestamp")
entry_date, entry_time = parse_date_time(timestamp_raw)
current_salary = cls._cell(data, "Current Salary")
expected_salary = cls._cell(data, "Expected Salary")
return {
"sheet": sheet,
"row_number": row_number,
"raw_record": dict(data),
"entry_year": entry_year,
"entry_date": entry_date,
"entry_time": entry_time,
"name": cls._cell(data, "Full Name"),
"gender": cls._cell(data, "Gender"),
"candidate_number": cls._cell(data, "Phone number (03XX-XXXXXXX)"),
"candidate_email": cls._cell(data, "Email"),
"date_of_birth": parse_date(data.get("Date of Birth")),
"cnic": cls._cell(data, "National Identification No. (42000-XXXXXXX-X)"),
"marital_status": cls._cell(data, "Marital Status"),
"position_suitable_for": cls._cell(data, "Position Applied For"),
"profile_link": cls._cell(data, "LinkedIn Profile Link"),
"residing_country": cls._cell(data, "Residing Country"),
"residing_city": cls._cell(data, "Residing City"),
"ho_availability": cls._cell(data, "Are you willing to relocate?"),
"degree": cls._cell(data, "Educational Degree"),
"university": cls._cell(data, "University"),
"university_other": cls._cell(
data,
"If your university is not listed above, please specify its name.",
),
"notice_period": cls._cell(data, "How soon can you join us?"),
"resume_link": cls._cell(data, "Drop your updated resume"),
"source_of_application": cls._cell(
data,
"Where did you hear about the position you're applying for?",
),
"cgpa": cls._cell(data, "CGPA"),
"area_of_expertise": cls._cell(data, "Area of Interest"),
"current_salary": current_salary,
"current_salary_value": parse_salary(current_salary),
"expected_salary": expected_salary,
"expected_salary_value": parse_salary(expected_salary),
"screened_by": cls._cell(data, "Recruiter"),
"hr_comments": cls._cell(data, "HR Comment"),
"director_poc_category": cls._cell(data, "Director / POC / Category"),
}
class SheetImportRun(SQLModel, table=True): class SheetImportRun(SQLModel, table=True):
"""One Google Sheet → FormData import job (Taskiq). Survives tab close.""" """One Google Sheet → FormData import job (Taskiq). Survives tab close."""

View File

@ -30,10 +30,7 @@ from g_sheet.plugins import (
build_sheets_client, build_sheets_client,
ensure_fresh, ensure_fresh,
execute, execute,
import_row_stats,
load_credentials, load_credentials,
map_record_to_form_data,
normalise_headers,
quote_tab, quote_tab,
rows_to_indexed_records, rows_to_indexed_records,
rows_to_records, rows_to_records,
@ -245,20 +242,17 @@ class SheetImport(SheetRead):
rows=data["rows"] rows=data["rows"]
if not rows: if not rows:
return serialize_import({"tab":tab,"rows_read":0,"inserted":0,"deleted":0}) return serialize_import({"tab":tab,"rows_read":0,"inserted":0,"deleted":0})
headers=normalise_headers(rows[0])
indexed=rows_to_indexed_records(rows) indexed=rows_to_indexed_records(rows)
mapped=[ mapped=[
map_record_to_form_data(tab,record,headers,row_number) FormData.from_sheet_row(tab,row_number,record)
for row_number,record in indexed for row_number,record in indexed
] ]
result=await FormData.replace_sheet(session,tab,mapped) result=await FormData.replace_sheet(session,tab,mapped)
stats=import_row_stats(mapped,headers)
return serialize_import({ return serialize_import({
"tab":tab, "tab":tab,
"rows_read":len(indexed), "rows_read":len(indexed),
"inserted":result["inserted"], "inserted":result["inserted"],
"deleted":result["deleted"], "deleted":result["deleted"],
**stats,
}) })
async def import_all(self): async def import_all(self):

View File

@ -1,6 +1,6 @@
"""Unit tests for g_sheet/plugins.py — mapping against the real 32-column sheet. """Unit tests for FormData.from_sheet_row + sheet row helpers.
No DB, no network. Pins header aliases, parsers, and the Bilal sample row. No DB, no network. Pins the Google Form Responses header keys and YoG rules.
""" """
from __future__ import annotations from __future__ import annotations
@ -8,154 +8,119 @@ from __future__ import annotations
from datetime import datetime, timezone from datetime import datetime, timezone
from g_sheet import plugins from g_sheet import plugins
from g_sheet.enums import FORM_DATA_FIELDS, HEADER_ALIASES, FormDataField from g_sheet.enums import FORM_DATA_FIELDS
from g_sheet.models import FormData from g_sheet.models import FormData
# Real header row from "PK - Recruitment Tracking Sheet" (32 columns). FORM_RESPONSE_RECORD = {
HEADERS = [ "Timestamp": "7/2/2026 17:50:19",
"um", "Email": "nusratazra@gmail.com",
"Year", "Full Name": "Nusrat Azra",
"Month", "Gender": "Female",
"Date", "Date of Birth": "7/24/1984",
"Time of Entry", "Marital Status": "Single",
"Screened By", "CGPA": "3.5",
"Candidate Name", "University": "Karachi University",
"HR comments", "If your university is not listed above, please specify its name.": "",
"Candidate Number", "Educational Degree": "Masters",
"Candidate Email", "Year of Graduation": "12/2/2007",
"Profile Link", "Position Applied For": "Executive Secretary",
"Area of Expertise", "LinkedIn Profile Link": (
"Requisition Number", "https://www.linkedin.com/in/nusrat-azra-executive-manager-to-c-suite-3bb86719/"
"Position Suitable For", ),
"Source of Application", "Drop your updated resume": "https://drive.google.com/open?id=1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4",
"Age", "How soon can you join us?": "1 - 2 weeks",
"Marital Status", "Phone number (03XX-XXXXXXX)": "03312464228",
"Education", "Are you willing to relocate?": "Yes",
"University of Graduation ", "Residing City": "Karachi",
"Experience", "Residing Country": "Pakistan",
"Experience Details", "Area of Interest": "",
"Area of Residence", "Recruiter": "",
"Communication Skills\n(01 to 10)", "Director / POC / Category": "Operations",
"Preferred Timings?", "National Identification No. (42000-XXXXXXX-X)": "4250105627772",
"Availability to work in the H.O", "Where did you hear about the position you're applying for?": "Indeed",
"Current Company",
"Reason for Leaving",
"How soon can you join",
"Current Salary",
"Expected Salary",
"Pros",
"Cons",
]
EXPECTED_FIELDS = (
FormDataField.SERIAL_NO,
FormDataField.ENTRY_YEAR,
FormDataField.ENTRY_MONTH,
FormDataField.ENTRY_DATE,
FormDataField.ENTRY_TIME,
FormDataField.SCREENED_BY,
FormDataField.NAME,
FormDataField.HR_COMMENTS,
FormDataField.CANDIDATE_NUMBER,
FormDataField.CANDIDATE_EMAIL,
FormDataField.PROFILE_LINK,
FormDataField.AREA_OF_EXPERTISE,
FormDataField.REQUISITION_NUMBER,
FormDataField.POSITION_SUITABLE_FOR,
FormDataField.SOURCE_OF_APPLICATION,
FormDataField.AGE,
FormDataField.MARITAL_STATUS,
FormDataField.DEGREE,
FormDataField.UNIVERSITY,
FormDataField.EXPERIENCE,
FormDataField.EXPERIENCE_DETAILS,
FormDataField.AREA_OF_RESIDENCE,
FormDataField.COMMUNICATION_SKILLS,
FormDataField.PREFERRED_TIMINGS,
FormDataField.HO_AVAILABILITY,
FormDataField.CURRENT_COMPANY,
FormDataField.REASON_FOR_LEAVING,
FormDataField.NOTICE_PERIOD,
FormDataField.CURRENT_SALARY,
FormDataField.EXPECTED_SALARY,
FormDataField.PROS,
FormDataField.CONS,
)
SAMPLE_RECORD = {
"um": "1",
"Year": "2021",
"Month": "June",
"Date": "23-Jun-2021",
"Time of Entry": "10:30 AM",
"Screened By": "Sara",
"Candidate Name": "Muhammad Bilal Khan",
"HR comments": "Good profile",
"Candidate Number": "0303-2892503",
"Candidate Email": "bilal_kf@yahoo.com",
"Profile Link": "https://example.com/bilal",
"Area of Expertise": "Software Development",
"Requisition Number": "REQ-1",
"Position Suitable For": "Backend Engineer",
"Source of Application": "Referral",
"Age": "33",
"Marital Status": "Married",
"Education": "BS CS",
"University of Graduation ": "NUST",
"Experience": "11 Years",
"Experience Details": "Software development related experience",
"Area of Residence": "Islamabad",
"Communication Skills\n(01 to 10)": "7",
"Preferred Timings?": "Morning",
"Availability to work in the H.O": "Yes",
"Current Company": "Acme",
"Reason for Leaving": "Growth",
"How soon can you join": "1 month",
"Current Salary": "110k", "Current Salary": "110k",
"Expected Salary": "150k", "Expected Salary": "150k",
"Pros": "Strong backend", "HR Comment": "Good profile",
"Cons": "Limited cloud",
} }
def test_every_real_header_maps_and_none_are_unmapped(): def test_from_sheet_row_maps_form_response_keys():
assert len(HEADERS) == 32 mapped = FormData.from_sheet_row(
for header, field in zip(HEADERS, EXPECTED_FIELDS): "Form Responses - Candidate Database Sheet 2026",
assert plugins.match_field(header) == field, header 9,
assert plugins.collect_unmapped_headers(HEADERS) == [] FORM_RESPONSE_RECORD,
)
assert mapped["name"] == "Nusrat Azra"
assert mapped["candidate_email"] == "nusratazra@gmail.com"
assert mapped["candidate_number"] == "03312464228"
assert mapped["gender"] == "Female"
assert mapped["date_of_birth"] == datetime(1984, 7, 24, tzinfo=timezone.utc)
assert mapped["cnic"] == "4250105627772"
assert mapped["cgpa"] == "3.5"
assert mapped["degree"] == "Masters"
assert mapped["university"] == "Karachi University"
assert mapped["position_suitable_for"] == "Executive Secretary"
assert mapped["notice_period"] == "1 - 2 weeks"
assert mapped["source_of_application"] == "Indeed"
assert mapped["ho_availability"] == "Yes"
assert mapped["marital_status"] == "Single"
assert mapped["residing_city"] == "Karachi"
assert mapped["residing_country"] == "Pakistan"
assert mapped["director_poc_category"] == "Operations"
assert mapped["hr_comments"] == "Good profile"
assert mapped["current_salary_value"] == 110000
assert mapped["expected_salary_value"] == 150000
assert mapped["entry_year"] == "12/2/2007"
assert mapped["entry_date"] is not None
assert mapped["entry_time"] == "17:50"
assert mapped["name"] != "7/2/2026 17:50:19"
assert mapped["row_number"] == 9
assert mapped["sheet"] == "Form Responses - Candidate Database Sheet 2026"
assert mapped["raw_record"]["Full Name"] == "Nusrat Azra"
def test_experience_details_does_not_overwrite_experience(): def test_year_of_graduation_prefers_second_when_present():
mapped = plugins.map_record_to_form_data("tab", SAMPLE_RECORD, HEADERS, 2) data = {
assert mapped["experience"] == "11 Years" "Full Name": "Ada",
assert mapped["experience_details"] == "Software development related experience" "Year of Graduation": "2010",
"Year of Graduation_1": "2015",
}
mapped = FormData.from_sheet_row("tab", 2, data)
assert mapped["entry_year"] == "2015"
def test_candidate_number_and_email_land_in_own_columns(): def test_year_of_graduation_uses_first_when_second_blank():
mapped = plugins.map_record_to_form_data("tab", SAMPLE_RECORD, HEADERS, 2) data = {
assert mapped["candidate_number"] == "0303-2892503" "Full Name": "Ada",
assert mapped["candidate_email"] == "bilal_kf@yahoo.com" "Year of Graduation": "2010",
assert mapped["name"] == "Muhammad Bilal Khan" "Year of Graduation_1": " ",
}
mapped = FormData.from_sheet_row("tab", 2, data)
assert mapped["entry_year"] == "2010"
def test_date_maps_to_entry_date_not_interview_round(): def test_year_of_graduation_none_when_both_blank():
mapped = plugins.map_record_to_form_data("tab", SAMPLE_RECORD, HEADERS, 2) data = {
assert mapped["entry_date"] == datetime(2021, 6, 23, tzinfo=timezone.utc) "Full Name": "Ada",
assert "interview_date" not in mapped "Year of Graduation": "",
"Year of Graduation_1": "",
}
mapped = FormData.from_sheet_row("tab", 2, data)
assert mapped["entry_year"] is None
def test_marital_status_only_lands_in_marital_status(): def test_year_of_graduation_single_column():
mapped = plugins.map_record_to_form_data("tab", SAMPLE_RECORD, HEADERS, 2) data = {"Full Name": "Ada", "Year of Graduation": "2012"}
assert mapped["marital_status"] == "Married" mapped = FormData.from_sheet_row("tab", 2, data)
assert "family_details" not in mapped assert mapped["entry_year"] == "2012"
assert "interview_status" not in mapped
def test_canonical_header_strips_parens_question_and_trailing_space(): def test_duplicate_year_header_becomes_year_of_graduation_1():
assert plugins.canonical_header("Communication Skills\n(01 to 10)") == "communication skills" headers = plugins.normalise_headers(
assert plugins.canonical_header("Preferred Timings?") == "preferred timings" ["Full Name", "Year of Graduation", "Year of Graduation"],
assert plugins.canonical_header("University of Graduation ") == "university of graduation" )
assert headers == ["Full Name", "Year of Graduation", "Year of Graduation_1"]
def test_parse_salary_and_score(): def test_parse_salary_and_score():
@ -178,131 +143,11 @@ def test_rows_to_indexed_records_keeps_true_sheet_row_across_blank():
(2, {"Name": "Ada", "Age": "30"}), (2, {"Name": "Ada", "Age": "30"}),
(4, {"Name": "Bob", "Age": "40"}), (4, {"Name": "Bob", "Age": "40"}),
] ]
# read API contract still drops blanks without exposing indices
assert plugins.rows_to_records(rows) == [ assert plugins.rows_to_records(rows) == [
{"Name": "Ada", "Age": "30"}, {"Name": "Ada", "Age": "30"},
{"Name": "Bob", "Age": "40"}, {"Name": "Bob", "Age": "40"},
] ]
def test_no_alias_string_appears_under_two_fields():
seen: dict[str, FormDataField] = {}
for field, aliases in HEADER_ALIASES.items():
for alias in aliases:
assert alias not in seen, f"{alias!r} under {seen[alias]} and {field}"
seen[alias] = field
def test_form_data_fields_match_model(): def test_form_data_fields_match_model():
assert set(FORM_DATA_FIELDS) == set(FormData.model_fields) assert set(FORM_DATA_FIELDS) == set(FormData.model_fields)
def test_sample_row_end_to_end():
mapped = plugins.map_record_to_form_data(
"PK - Recruitment Tracking Sheet", SAMPLE_RECORD, HEADERS, 2,
)
assert mapped["experience"] == "11 Years"
assert mapped["experience_details"] == "Software development related experience"
assert mapped["candidate_email"] == "bilal_kf@yahoo.com"
assert mapped["age"] == 33
assert mapped["current_salary_value"] == 110000
assert mapped["communication_skills"] == 7
assert mapped["pros"] == "Strong backend"
assert mapped["cons"] == "Limited cloud"
assert mapped["row_number"] == 2
assert mapped["sheet"] == "PK - Recruitment Tracking Sheet"
assert mapped.get("job_post_id") is None
# Google Form Responses tab — headers differ from the PK screening sheet.
FORM_RESPONSE_HEADERS = [
"Timestamp",
"Email",
"Full Name",
"Gender",
"Date of Birth",
"Marital Status",
"CGPA",
"University",
"If your university is not listed above, please specify its name.",
"Educational Degree",
"Year of Graduation",
"Position Applied For",
"LinkedIn Profile Link",
"Drop your updated resume",
"How soon can you join us?",
"Phone number (03XX-XXXXXXX)",
"Are you willing to relocate?",
"Residing City",
"Residing Country",
"Area of Interest",
"Recruiter",
"Director / POC / Category",
"National Identification No. (42000-XXXXXXX-X)",
"Where did you hear about the position you're applying for?",
]
FORM_RESPONSE_RECORD = {
"Timestamp": "7/2/2026 17:50:19",
"Email": "nusratazra@gmail.com",
"Full Name": "Nusrat Azra",
"Gender": "Female",
"Date of Birth": "7/24/1984",
"Marital Status": "Single",
"CGPA": "3.5",
"University": "Karachi University",
"If your university is not listed above, please specify its name.": "",
"Educational Degree": "Masters",
"Year of Graduation": "12/2/2007",
"Position Applied For": "Executive Secretary",
"LinkedIn Profile Link": "https://www.linkedin.com/in/nusrat-azra-executive-manager-to-c-suite-3bb86719/",
"Drop your updated resume": "https://drive.google.com/open?id=1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4",
"How soon can you join us?": "1 - 2 weeks",
"Phone number (03XX-XXXXXXX)": "03312464228",
"Are you willing to relocate?": "Yes",
"Residing City": "Karachi",
"Residing Country": "Pakistan",
"Area of Interest": "",
"Recruiter": "",
"Director / POC / Category": "Operations",
"National Identification No. (42000-XXXXXXX-X)": "4250105627772",
"Where did you hear about the position you're applying for?": "Indeed",
}
def test_form_response_headers_map_to_typed_columns():
assert plugins.collect_unmapped_headers(FORM_RESPONSE_HEADERS) == []
mapped = plugins.map_record_to_form_data(
"Form Responses - Candidate Database Sheet 2026",
FORM_RESPONSE_RECORD,
FORM_RESPONSE_HEADERS,
9,
)
assert mapped["name"] == "Nusrat Azra"
assert mapped["candidate_email"] == "nusratazra@gmail.com"
assert mapped["candidate_number"] == "03312464228"
assert mapped["gender"] == "Female"
assert mapped["date_of_birth"] == datetime(1984, 7, 24, tzinfo=timezone.utc)
assert mapped["cnic"] == "4250105627772"
assert mapped["cgpa"] == "3.5"
assert mapped["degree"] == "Masters"
assert mapped["university"] == "Karachi University"
assert mapped["position_suitable_for"] == "Executive Secretary"
assert mapped["notice_period"] == "1 - 2 weeks"
assert mapped["source_of_application"] == "Indeed"
assert mapped["ho_availability"] == "Yes"
assert mapped["marital_status"] == "Single"
assert mapped["residing_city"] == "Karachi"
assert mapped["residing_country"] == "Pakistan"
assert mapped["director_poc_category"] == "Operations"
assert mapped["profile_link"] == (
"https://www.linkedin.com/in/nusrat-azra-executive-manager-to-c-suite-3bb86719/"
)
assert mapped["resume_link"] == "https://drive.google.com/open?id=1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4"
assert mapped["entry_year"] == "12/2/2007"
assert mapped["entry_date"] is not None
assert mapped["entry_time"] == "17:50"
# Timestamp must never be used as the candidate name.
assert mapped["name"] != "7/2/2026 17:50:19"
assert mapped["job_post_id"] is None
assert mapped["raw_record"]["Full Name"] == "Nusrat Azra"