# mapping_rules.py import re, datetime KNOWN_HEADWORKS = ["Tarbela","Kalabagh","Chashma","Taunsa","Taunsa","Guddu","Sukkur","Kotri","Mangla","Rasul","Marala", "Khanki","Qadirabad","Trimmu","Panjnad","Balloki","Sidhnai","Sulemanki","Islam",] def parse_number(s): if s is None: return None s = str(s).strip() if s == "": return None s = s.replace(',', '').replace('O','0').replace('o','0').replace('I','1').replace('l','1').replace('S','5') neg = False if re.search(r'^\(.*\)$', s): neg = True s = s.strip('()') m = re.search(r'[-+]?\d+(\.\d+)?', s) if not m: return None try: val = float(m.group(0)) except: return None return -val if neg else val def parse_date(s): if not s: return None s = str(s).strip() for fmt in ("%d-%m-%Y","%d/%m/%Y","%Y-%m-%d","%d-%m-%y","%d %b %Y","%d %B %Y"): try: return datetime.datetime.strptime(s, fmt).date().isoformat() except: pass m = re.search(r'(\d{1,2})[\/\-](\d{1,2})[\/\-](\d{2,4})', s) if m: d,mo,y = m.groups() if len(y)==2: y = '20'+y return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}" return None def find_known_headworks(text): if not text: return None for h in KNOWN_HEADWORKS: if h.lower() in text.lower(): return h return None def detect_unit(text): if not text: return None t = str(text) if re.search(r'\b(cs|Cs|CS)\b', t): return 'Cs' if re.search(r'\b(ft|FT)\b', t): return 'ft' if re.search(r'\b(ind|IND)\b', t): return 'IND' return None def map_row(cells, confidences=None, header_tokens=None): """ cells: list of strings (OCR per column) confidences: list of floats (0..1) header_tokens: optional list of header strings if row is header returns: dict with canonical fields """ if confidences is None: confidences = [0.95]*len(cells) joined = " | ".join([c for c in cells if c]) header = " ".join(header_tokens).lower() if header_tokens else "" out = { "Date": None, "Headworks": None, "Canal": None, "Site": None, "A.F.S.Q": None, "Gauge": None, "Indent": None, "Actual": None, "Unit": None, "SourceRowText": joined, "OCRConfidence": round(sum(confidences)/len(confidences),3) } # Date for c in cells: d = parse_date(c) if d: out["Date"] = d break # Headworks if header and "headwork" in header: for t in header_tokens: h = find_known_headworks(t) if h: out["Headworks"] = h break if not out["Headworks"]: for c in cells[:3]: h = find_known_headworks(c) if h: out["Headworks"] = h break # Gauge detection (decimal-like) gauge_idx = None for i,c in enumerate(cells): if c and re.search(r'\d+\.\d{1,2}', str(c)): gauge_idx = i out["Gauge"] = parse_number(c) break # Numeric candidates nums = [(i, parse_number(c)) for i,c in enumerate(cells) if parse_number(c) is not None] nums = [(i,v) for (i,v) in nums if i!=gauge_idx] nums_sorted = sorted(nums, key=lambda x: abs(x[1]) if x[1] is not None else 0, reverse=True) if len(nums_sorted)>=1: out["Indent"] = nums_sorted[0][1] if len(nums_sorted)>=2: out["Actual"] = nums_sorted[1][1] # A.F.S.Q detection (small decimal or single-digit) for c in cells: try: val = float(str(c).strip()) if val>0 and val<30: out["A.F.S.Q"] = val break except: continue # Unit detection for c in cells: u = detect_unit(c) if u: out["Unit"] = u break # Canal and Site heuristics for c in cells: if c and any(k in str(c).lower() for k in ["canal","link","feeder","escape","branch","rd","rd-","tail","head"]): if not out["Canal"]: out["Canal"] = str(c).strip().title() elif not out["Site"]: out["Site"] = str(c).strip().upper() # Site tokens U/S D/S HEAD TAIL for c in cells: if c and any(k in str(c).upper() for k in ["U/S","D/S","HEAD","TAIL","RD","RD-"]): out["Site"] = str(c).strip().upper() break return out