|
|
|
|
| import os
|
| import re
|
| import csv
|
| import json
|
| import ftfy
|
| from datetime import datetime
|
| from dateutil.parser import parse as date_parse
|
|
|
| from schemas import DEFAULT_BLANK_FIELDS, CHOICE_OPTIONS
|
| from config import MIN_CONFIDENCE_THRESHOLD
|
| from agent import Field
|
|
|
| def serialize_knowledge_base(knowledge_base: dict) -> dict:
|
| serializable_kb = {}
|
| for variant, data in knowledge_base.items():
|
| serializable_kb[variant] = {field_name: field_obj.to_dict() for field_name, field_obj in data.items()}
|
| return serializable_kb
|
|
|
| def deserialize_knowledge_base(cached_data: dict) -> dict:
|
| knowledge_base = {}
|
| for variant, data in cached_data.items():
|
| knowledge_base[variant] = {
|
| field_name: Field.from_dict(field_data) for field_name, field_data in data.items()
|
| }
|
| return knowledge_base
|
|
|
| class SafeCSVWriter:
|
| def __init__(self, filepath, schema):
|
| self.filepath = filepath
|
| self.schema = schema
|
| self.file = None
|
| self.writer = None
|
|
|
| def __enter__(self):
|
| os.makedirs(os.path.dirname(self.filepath), exist_ok=True)
|
| file_exists = os.path.isfile(self.filepath) and os.path.getsize(self.filepath) > 0
|
| self.file = open(self.filepath, 'a', newline='', encoding='utf-8')
|
| self.writer = csv.DictWriter(self.file, fieldnames=self.schema)
|
|
|
| if not file_exists:
|
| self.writer.writeheader()
|
| self.file.flush()
|
| return self
|
|
|
| def writerow(self, row):
|
| if self.writer:
|
| self.writer.writerow(row)
|
| self.file.flush()
|
|
|
| def __exit__(self, exc_type, exc_val, exc_tb):
|
| if self.file:
|
| self.file.close()
|
|
|
| def normalize_restricted_traffic(value: str) -> str:
|
| if not value: return "No"
|
| v = value.lower().strip()
|
| if v in['yes', 'true', '1', 'restricted']: return "Yes"
|
| return "No"
|
|
|
| def validate_row_integrity(row: dict, schema: list) -> bool:
|
| if 'event' in schema and not row.get('event'):
|
| print(f"[VALIDATION FAIL] Missing 'event' field.")
|
| return False
|
| if row.get('date'):
|
| if len(row['date']) > 50 or "http" in row['date']:
|
| row['date'] = ""
|
| return True
|
|
|
| def format_cutoff_time(val: str) -> str:
|
| """Strictly parses cutoff text into HH:MM format."""
|
| val = str(val).lower().strip()
|
| if not val or val in['na', 'n/a', 'none', 'null', 'unknown', '']:
|
| return ""
|
|
|
|
|
| m = re.search(r'\b(\d{1,2}):(\d{2})(?::\d{2})?\b', val)
|
| if m:
|
| return f"{int(m.group(1)):02d}:{m.group(2)}"
|
|
|
|
|
| m = re.search(r'(\d+\.\d+)\s*(?:h|hr|hour)', val)
|
| if m:
|
| h_float = float(m.group(1))
|
| h = int(h_float)
|
| mins = int(round((h_float - h) * 60))
|
| return f"{h:02d}:{mins:02d}"
|
|
|
|
|
| m = re.search(r'(\d+)\s*(?:h|hr|hour)[^0-9]*(\d+)\s*(?:m|min)', val)
|
| if m:
|
| return f"{int(m.group(1)):02d}:{int(m.group(2)):02d}"
|
|
|
|
|
| m = re.search(r'\b(\d+)\s*(?:h|hr|hour)\b', val)
|
| if m:
|
| return f"{int(m.group(1)):02d}:00"
|
|
|
| return ""
|
|
|
| def format_final_row(festival_name_input: str, variant_name: str, data: dict, schema: list) -> dict | None:
|
|
|
| def normalize_complex_value(val):
|
| if isinstance(val, str):
|
| val = val.strip()
|
|
|
| if val.startswith("AttributedDict"): return ""
|
| if val.startswith('{') and val.endswith('}'):
|
| try: val = json.loads(val)
|
| except json.JSONDecodeError: return val
|
| elif val.startswith('[') and val.endswith(']'):
|
| try: val = json.loads(val)
|
| except json.JSONDecodeError: return val
|
| else:
|
| return val
|
|
|
| if isinstance(val, dict):
|
| for key in['amount', 'value', 'price', 'final', 'standard', 'date', 'time', 'name']:
|
| if key in val: return normalize_complex_value(val[key])
|
| parts =[]
|
| for v in val.values():
|
| if isinstance(v, (int, float, str, bool)):
|
| parts.append(str(v))
|
| return " | ".join(parts) if parts else ""
|
|
|
| if isinstance(val, list):
|
| strings =[str(i) for i in val if isinstance(i, (int, float, str, bool))]
|
| return ", ".join(strings) if strings else ""
|
|
|
| return str(val) if val is not None else ""
|
|
|
| def get_value(field_name: str) -> any:
|
| field_obj = data.get(field_name)
|
| if not isinstance(field_obj, Field): return ""
|
| if field_obj.inferred_by == 'pre_processed_data':
|
| return field_obj.value
|
| min_thresh = 0.4 if "inference" in field_obj.inferred_by else MIN_CONFIDENCE_THRESHOLD
|
| if field_obj.confidence >= min_thresh:
|
| return field_obj.value
|
| return ""
|
|
|
| def finalize_value(value: any) -> str:
|
| value = normalize_complex_value(value)
|
| text = ftfy.fix_text(str(value)).strip()
|
|
|
| if "AttributedDict" in text or "{" in text or "}" in text: return ""
|
| if text.lower() in["na", "n/a", "", "none", "not specified", "null", "unknown"]: return ""
|
| return text
|
|
|
| row = {}
|
| for key in schema:
|
| if key in DEFAULT_BLANK_FIELDS:
|
| row[key] = ""
|
| continue
|
|
|
| raw_value = finalize_value(get_value(key))
|
| if not raw_value:
|
| row[key] = ""
|
| continue
|
|
|
| if key == "city":
|
| row[key] = raw_value.split(',')[0].strip()
|
|
|
| elif key == "registrationCost":
|
| cleaned = re.sub(r'[^\d.]', '', raw_value)
|
| row[key] = cleaned
|
|
|
| elif key in["runningDistance", "cyclingDistance", "swimDistance"]:
|
| match = re.search(r'(\d+\.?\d*)', raw_value)
|
| row[key] = match.group(1) if match else ""
|
|
|
| elif key in["date", "lastDate"]:
|
| if re.match(r"\d{4}-\d{2}-\d{2}", raw_value):
|
| try: row[key] = datetime.strptime(raw_value, "%Y-%m-%d").strftime("%d-%m-%Y")
|
| except ValueError: row[key] = ""
|
| elif re.match(r"\d{2}-\d{2}-\d{4}", raw_value):
|
| row[key] = raw_value
|
| else:
|
| try:
|
| dt = date_parse(raw_value, fuzzy=True, dayfirst=True)
|
| if 2020 <= dt.year <= 2030: row[key] = dt.strftime("%d-%m-%Y")
|
| else: row[key] = ""
|
| except (ValueError, TypeError, OverflowError): row[key] = ""
|
|
|
| elif key == "startTime":
|
| try:
|
| dt = date_parse(raw_value, fuzzy=True)
|
| row[key] = dt.strftime("%H:%M")
|
| except (ValueError, TypeError, OverflowError):
|
| row[key] = ""
|
|
|
| elif "Cutoff" in key:
|
| row[key] = format_cutoff_time(raw_value)
|
|
|
| elif key in["firstEdition", "lastEdition", "editionYear"]:
|
| match = re.search(r'\b((?:19|20)\d{2})\b', raw_value)
|
| row[key] = match.group(1) if match else ""
|
|
|
| elif key == "restrictedTraffic":
|
| row[key] = normalize_restricted_traffic(raw_value)
|
|
|
| else:
|
| row[key] = raw_value
|
|
|
| if festival_name_input and festival_name_input.lower() not in variant_name.lower():
|
| row["event"] = f"{festival_name_input} - {variant_name}"
|
| else:
|
| row["event"] = variant_name
|
|
|
| if not row.get("organiser") and festival_name_input:
|
| row["organiser"] = festival_name_input
|
|
|
| if row.get("date"):
|
| try:
|
| dt = datetime.strptime(row["date"], "%d-%m-%Y")
|
| row["month"] = dt.strftime("%B")
|
| if not row.get("editionYear"):
|
| row["editionYear"] = str(dt.year)
|
| except Exception: pass
|
|
|
| if not row.get("approvalStatus"):
|
| row["approvalStatus"] = "Approved"
|
|
|
| if not validate_row_integrity(row, schema):
|
| return None
|
|
|
| return row |
