feat: support affiliation stints and improve extraction coverage

Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>

Author: skwowet

services/apps/git_integration/src/crowdgit/database/crud.py

@@ -726,60 +726,98 @@ async def insert_member_organizations(rows: list[dict]) -> None:
     if not rows:
         return
 
-    sql_query = """
+    undated_rows: list[tuple] = []
+    open_ended_rows: list[tuple] = []
+    dated_rows: list[tuple] = []
+
+    for row in rows:
+        params = (
+            row["member_id"],
+            row["organization_id"],
+            row.get("date_start"),
+            row.get("date_end"),
+            row["source"],
+        )
+        date_start = row.get("date_start")
+        date_end = row.get("date_end")
+        if date_start is None and date_end is None:
+            undated_rows.append(params)
+        elif date_end is None:
+            open_ended_rows.append(params)
+        else:
+            dated_rows.append(params)
+
+    insert_sql = """
         INSERT INTO "memberOrganizations"(
             "memberId",
             "organizationId",
             "dateStart",
             "dateEnd",
-            "title",
+            title,
             source,
-            verified,
             "createdAt",
             "updatedAt"
         )
-        VALUES ($1, $2, NULL, NULL, NULL, $3, false, NOW(), NOW())
-        ON CONFLICT ("memberId", "organizationId")
-            WHERE ("dateStart" IS NULL AND "dateEnd" IS NULL)
-        DO NOTHING
+        VALUES ($1, $2, $3, $4, NULL, $5, NOW(), NOW())
     """
-    await executemany(
-        sql_query,
-        [
-            (
-                row["member_id"],
-                row["organization_id"],
-                row.get("source", "project-registry"),
-            )
-            for row in rows
-        ],
-    )
+
+    if undated_rows:
+        sql = (
+            insert_sql
+            + """
+            ON CONFLICT ("memberId", "organizationId")
+                WHERE ("dateStart" IS NULL AND "dateEnd" IS NULL AND "deletedAt" IS NULL)
+            DO NOTHING
+        """
+        )
+        await executemany(sql, undated_rows)
+
+    if open_ended_rows:
+        sql = (
+            insert_sql
+            + """
+            ON CONFLICT ("memberId", "organizationId", "dateStart")
+                WHERE ("dateEnd" IS NULL AND "deletedAt" IS NULL)
+            DO NOTHING
+        """
+        )
+        await executemany(sql, open_ended_rows)
+
+    if dated_rows:
+        sql = (
+            insert_sql
+            + """
+            ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd")
+                WHERE ("deletedAt" IS NULL)
+            DO NOTHING
+        """
+        )
+        await executemany(sql, dated_rows)
 
 
 async def insert_member_segment_affiliations(rows: list[dict]) -> None:
     if not rows:
         return
 
-    sql_query = """
+    await executemany(
+        """
         INSERT INTO "memberSegmentAffiliations"(
             id,
             "memberId",
             "segmentId",
             "organizationId",
             "dateStart",
-            "dateEnd",
-            verified
+            "dateEnd"
         )
-        VALUES (gen_random_uuid(), $1, $2, $3, NULL, NULL, $4)
-    """
-    await executemany(
-        sql_query,
+        VALUES (gen_random_uuid(), $1, $2, $3, $4, $5)
+        """,
         [
             (
                 row["member_id"],
                 row["segment_id"],
                 row["organization_id"],
-                row.get("verified", False),
+                row.get("date_start"),
+                row.get("date_end"),
             )
             for row in rows
         ],

services/apps/git_integration/src/crowdgit/models/affiliation_info.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
 import uuid
-from datetime import datetime
+from datetime import date, datetime
 from typing import Any
 
 import orjson
 from loguru import logger
-from pydantic import BaseModel, TypeAdapter, ValidationError
+from pydantic import BaseModel, Field, TypeAdapter, ValidationError
 
 
 class AffiliationContributor(BaseModel):
@@ -15,14 +15,36 @@ class AffiliationContributor(BaseModel):
     github: str | None = None
 
 
-class AffiliationOrganization(BaseModel):
+class AffiliationOrganizationFields(BaseModel):
+    """Organization fields as returned by the parse AI (flat rows)."""
+
     name: str | None = None
     domain: str | None = None
+    date_start: date | None = Field(default=None, alias="dateStart")
+    date_end: date | None = Field(default=None, alias="dateEnd")
+    is_unaffiliated: bool = Field(default=False, alias="isUnaffiliated")
+
+    model_config = {"populate_by_name": True}
+
+
+class AffiliationParseRow(BaseModel):
+    contributor: AffiliationContributor
+    organization: AffiliationOrganizationFields
+
+
+class AffiliationOrganizationStint(BaseModel):
+    name: str | None = None
+    domain: str
+    date_start: date | None = Field(default=None, alias="dateStart")
+    date_end: date | None = Field(default=None, alias="dateEnd")
+    is_unaffiliated: bool = Field(default=False, alias="isUnaffiliated")
+
+    model_config = {"populate_by_name": True}
 
 
-class AffiliationInfoItem(BaseModel):
+class AffiliationContributorEntry(BaseModel):
     contributor: AffiliationContributor
-    organization: AffiliationOrganization
+    organizations: list[AffiliationOrganizationStint]
 
 
 class AffiliationFile(BaseModel):
@@ -31,19 +53,19 @@ class AffiliationFile(BaseModel):
 
 
 class AffiliationParseOutput(BaseModel):
-    affiliations: list[AffiliationInfoItem] | None = None
+    affiliations: list[AffiliationParseRow] | None = None
     error: str | None = None
 
 
-_SNAPSHOT_ADAPTER = TypeAdapter(list[AffiliationInfoItem])
+_SNAPSHOT_ADAPTER = TypeAdapter(list[AffiliationContributorEntry])
 
 
 class RepoAffiliationRegistry(BaseModel):
     repo_id: str
     file_path: str | None = None
     file_hash: str | None = None
     status: str
-    snapshot: list[AffiliationInfoItem] | None = None
+    snapshot: list[AffiliationContributorEntry] | None = None
     last_run_at: datetime | None = None
 
     @classmethod
@@ -71,7 +93,7 @@ def from_db(cls, db_data: dict[str, Any]) -> RepoAffiliationRegistry:
         return cls(**row)
 
     @staticmethod
-    def _parse_snapshot(snapshot) -> list[AffiliationInfoItem] | None:
+    def _parse_snapshot(snapshot) -> list[AffiliationContributorEntry] | None:
         if isinstance(snapshot, str | bytes):
             try:
                 snapshot = orjson.loads(snapshot)
@@ -91,4 +113,6 @@ def _parse_snapshot(snapshot) -> list[AffiliationInfoItem] | None:
     def snapshot_for_db(self) -> str | None:
         if self.snapshot is None:
             return None
-        return orjson.dumps([item.model_dump() for item in self.snapshot]).decode()
+        return orjson.dumps(
+            [item.model_dump(mode="json", by_alias=True) for item in self.snapshot]
+        ).decode()