451 lines
18 KiB
Python
451 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
import os
|
|
import json
|
|
import uuid
|
|
import re
|
|
from datetime import datetime
|
|
|
|
# Import normalization helpers from extract.py
|
|
import extract as ex
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
OUTPUT_DIR = os.path.join(HERE, "output")
|
|
SEEDS_PATH = os.path.join(HERE, "..", "..", "gerbil-manager-web", "src", "genetics", "colorVarietySeed.backend.json")
|
|
|
|
def generate_guid(key_str):
|
|
"""Generate a stable UUID string based on a key."""
|
|
return str(uuid.uuid5(uuid.NAMESPACE_DNS, key_str))
|
|
|
|
def parse_date_only(d):
|
|
"""Convert DD.MM.YYYY string to YYYY-MM-DD for JSON serialization."""
|
|
if not d:
|
|
return None
|
|
try:
|
|
p = d.split(".")
|
|
if len(p) == 3:
|
|
day = int(p[0])
|
|
month = int(p[1])
|
|
year = int(p[2])
|
|
if len(p[2]) == 2:
|
|
year += 2000
|
|
# Ensure valid date
|
|
dt = datetime(year, month, day)
|
|
return dt.strftime("%Y-%m-%d")
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
def save_to_csv(data_list, filepath):
|
|
import csv
|
|
if not data_list:
|
|
return
|
|
headers = list(data_list[0].keys())
|
|
with open(filepath, "w", encoding="utf-8-sig", newline="") as f:
|
|
writer = csv.writer(f, delimiter=";")
|
|
writer.writerow(headers)
|
|
for row in data_list:
|
|
values = []
|
|
for h in headers:
|
|
val = row[h]
|
|
if isinstance(val, (list, dict)):
|
|
values.append(json.dumps(val, ensure_ascii=False))
|
|
elif val is None:
|
|
values.append("")
|
|
else:
|
|
values.append(str(val))
|
|
writer.writerow(values)
|
|
|
|
def main():
|
|
print("Loading extracted JSON files...")
|
|
animals_file = os.path.join(OUTPUT_DIR, "animals.json")
|
|
litters_file = os.path.join(OUTPUT_DIR, "litters.json")
|
|
|
|
if not os.path.exists(animals_file) or not os.path.exists(litters_file):
|
|
print(f"Error: extract.py must be run first to generate animals.json and litters.json in {OUTPUT_DIR}")
|
|
return
|
|
|
|
with open(animals_file, "r", encoding="utf-8") as f:
|
|
animals = json.load(f)
|
|
with open(litters_file, "r", encoding="utf-8") as f:
|
|
litters = json.load(f)
|
|
|
|
# Load color variety seeds
|
|
print("Loading color variety seeds...")
|
|
variety_map = {}
|
|
|
|
# Load from C# ApplicationContext.cs catalog for stable database GUIDs (index + 1)
|
|
app_context_path = os.path.abspath(os.path.join(HERE, "..", "..", "GerbilManagerWebAPI", "ApplicationContext.cs"))
|
|
cs_name_to_id = {}
|
|
if os.path.exists(app_context_path):
|
|
with open(app_context_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
catalog_match = re.search(r'catalog\s*=\s*\{(.*?)\};', content, re.DOTALL)
|
|
if catalog_match:
|
|
block = catalog_match.group(1)
|
|
entries = re.findall(r'\(\s*"([^"]+)"\s*,\s*"([^"]+)"\s*,\s*(\d+)\s*\)', block)
|
|
for idx, (name, genotype, sort_order) in enumerate(entries):
|
|
variety_id = f"00000000-0000-0000-0000-{idx + 1:012d}"
|
|
cs_name_to_id[name.strip().lower()] = variety_id
|
|
variety_map[name.strip().lower()] = variety_id
|
|
else:
|
|
print(f"Warning: ApplicationContext.cs not found at {app_context_path}")
|
|
|
|
with open(SEEDS_PATH, "r", encoding="utf-8") as f:
|
|
seeds = json.load(f)
|
|
|
|
for v in seeds:
|
|
name_lower = v["name"].strip().lower()
|
|
variety_id = cs_name_to_id.get(name_lower)
|
|
if not variety_id:
|
|
variety_id = f"00000000-0000-0000-0000-{v['sortOrder'] + 1:012d}"
|
|
variety_map[name_lower] = variety_id
|
|
|
|
# Map English name if present
|
|
if "english" in v and v["english"]:
|
|
variety_map[v["english"].strip().lower()] = variety_id
|
|
|
|
# 1. Establish stable Guid maps
|
|
animal_guid_map = {a["id"]: generate_guid(f"animal-{a['id']}") for a in animals}
|
|
litter_guid_map = {l["id"]: generate_guid(f"litter-{l['id']}") for l in litters}
|
|
|
|
# Index animals for fast lookup by slug ID and by normalized name+dob
|
|
animal_by_id = {a["id"]: a for a in animals}
|
|
animal_by_name_dob = {}
|
|
animal_by_name = {}
|
|
for a in animals:
|
|
key = (ex.norm_name(a["name"]), ex.norm_dob(a["dob"]))
|
|
if key[0] and key[1]:
|
|
animal_by_name_dob.setdefault(key, []).append(a)
|
|
nk = ex.norm_name(a["name"])
|
|
if nk:
|
|
animal_by_name.setdefault(nk, []).append(a)
|
|
|
|
|
|
# 2. Extract and resolve unique Contacts (Breeders & Zuchten)
|
|
print("Extracting unique contacts...")
|
|
contact_names = set()
|
|
for a in animals:
|
|
if a.get("breeder"):
|
|
contact_names.add(a["breeder"].strip())
|
|
if a.get("zucht"):
|
|
contact_names.add(a["zucht"].strip())
|
|
for l in litters:
|
|
if l.get("damZucht"):
|
|
contact_names.add(l["damZucht"].strip())
|
|
if l.get("sireZucht"):
|
|
contact_names.add(l["sireZucht"].strip())
|
|
|
|
contact_guid_map = {}
|
|
resolved_contacts = []
|
|
for idx, name in enumerate(sorted(contact_names)):
|
|
norm = name.strip().lower()
|
|
if not norm or norm in contact_guid_map:
|
|
continue
|
|
c_guid = generate_guid(f"contact-{norm}")
|
|
contact_guid_map[norm] = c_guid
|
|
resolved_contacts.append({
|
|
"Id": c_guid,
|
|
"Name": name,
|
|
"ZuchtName": name if "zucht" in norm or "clan" in norm or "runner" in norm else "",
|
|
"City": "",
|
|
"Email": "",
|
|
"Homepage": "",
|
|
"Phone": "",
|
|
"Address": ""
|
|
})
|
|
|
|
# 3. Resolve parent relationships for litters
|
|
print("Resolving litter parents...")
|
|
litter_parent_map = {} # litter_id -> (father_guid, mother_guid)
|
|
|
|
# Pre-map parents using offspring parentRefs
|
|
for a in animals:
|
|
litter_ref = a.get("litterRef")
|
|
if not litter_ref or not litter_ref.get("litterId"):
|
|
continue
|
|
|
|
l_id = litter_ref["litterId"]
|
|
l_parents = litter_parent_map.setdefault(l_id, [None, None]) # [father, mother]
|
|
|
|
for p_ref in a.get("parentRefs", []):
|
|
p_dob = ex.norm_dob(p_ref.get("dob", ""))
|
|
p_key = (ex.norm_name(p_ref["name"]), p_dob)
|
|
p_candidates = animal_by_name_dob.get(p_key, [])
|
|
if not p_candidates:
|
|
# Fallback to name-only lookup when parent DOB is empty/not found
|
|
p_norm = ex.norm_name(p_ref["name"])
|
|
candidates = animal_by_name.get(p_norm, [])
|
|
if candidates:
|
|
offspring_dob_str = parse_date_only(a["dob"])
|
|
if offspring_dob_str:
|
|
try:
|
|
o_dob = datetime.strptime(offspring_dob_str, "%Y-%m-%d")
|
|
valid_candidates = []
|
|
for cand in candidates:
|
|
cand_dob_str = parse_date_only(cand["dob"])
|
|
if cand_dob_str:
|
|
try:
|
|
c_dob = datetime.strptime(cand_dob_str, "%Y-%m-%d")
|
|
if c_dob < o_dob:
|
|
valid_candidates.append(cand)
|
|
except ValueError:
|
|
valid_candidates.append(cand)
|
|
else:
|
|
valid_candidates.append(cand)
|
|
if valid_candidates:
|
|
p_candidates = [valid_candidates[0]]
|
|
except ValueError:
|
|
p_candidates = [candidates[0]]
|
|
else:
|
|
p_candidates = [candidates[0]]
|
|
if p_candidates:
|
|
p_guid = animal_guid_map[p_candidates[0]["id"]]
|
|
if p_ref["roleGuess"] == "father":
|
|
l_parents[0] = p_guid
|
|
elif p_ref["roleGuess"] == "mother":
|
|
l_parents[1] = p_guid
|
|
|
|
# Fallback lookup for parents by name from litter details if offspring has no parentRefs
|
|
for l in litters:
|
|
l_parents = litter_parent_map.setdefault(l["id"], [None, None])
|
|
l_date_str = parse_date_only(l["date"])
|
|
if not l_date_str:
|
|
continue
|
|
l_date = datetime.strptime(l_date_str, "%Y-%m-%d")
|
|
|
|
# Fallback Father
|
|
if not l_parents[0] and l.get("sireName"):
|
|
sire_norm = ex.norm_name(l["sireName"])
|
|
# Find an animal with this name born before the litter
|
|
candidates = []
|
|
for (name_norm, dob_norm), grp in animal_by_name_dob.items():
|
|
if name_norm == sire_norm:
|
|
for cand in grp:
|
|
c_dob_str = parse_date_only(cand["dob"])
|
|
if c_dob_str:
|
|
c_dob = datetime.strptime(c_dob_str, "%Y-%m-%d")
|
|
if c_dob < l_date:
|
|
candidates.append(cand)
|
|
if len(candidates) == 1:
|
|
l_parents[0] = animal_guid_map[candidates[0]["id"]]
|
|
|
|
# Fallback Mother
|
|
if not l_parents[1] and l.get("damName"):
|
|
dam_norm = ex.norm_name(l["damName"])
|
|
candidates = []
|
|
for (name_norm, dob_norm), grp in animal_by_name_dob.items():
|
|
if name_norm == dam_norm:
|
|
for cand in grp:
|
|
c_dob_str = parse_date_only(cand["dob"])
|
|
if c_dob_str:
|
|
c_dob = datetime.strptime(c_dob_str, "%Y-%m-%d")
|
|
if c_dob < l_date:
|
|
candidates.append(cand)
|
|
if len(candidates) == 1:
|
|
l_parents[1] = animal_guid_map[candidates[0]["id"]]
|
|
|
|
# 4. Construct Litters collection
|
|
resolved_litters = []
|
|
for l in litters:
|
|
l_guid = litter_guid_map[l["id"]]
|
|
p_father, p_mother = litter_parent_map.get(l["id"], [None, None])
|
|
l_date = parse_date_only(l["date"])
|
|
if not l_date:
|
|
continue # Skip litters without dates
|
|
|
|
# Inzucht/Frühsterblichkeit mapping
|
|
deaths_8w = None
|
|
if l.get("diedLater") is not None or l.get("stillborn") is not None:
|
|
deaths_8w = (l.get("stillborn") or 0) + (l.get("diedLater") or 0)
|
|
|
|
resolved_litters.append({
|
|
"Id": l_guid,
|
|
"Name": f"Wurf {l['litterId']}".strip(),
|
|
"Date": l_date,
|
|
"TotalBorn": l.get("totalBorn"),
|
|
"DeathsWithin8Weeks": deaths_8w,
|
|
"FatherId": p_father,
|
|
"MotherId": p_mother,
|
|
"ExpectedGoHomeDate": None,
|
|
"Notes": l.get("note") if l.get("note") else None,
|
|
"PairingCode": l.get("zuchtnummer") if l.get("zuchtnummer") else None,
|
|
"ExternalRef": l["id"],
|
|
"LitterLetter": l["litterId"] if len(l["litterId"]) <= 2 else None
|
|
})
|
|
|
|
# 5. Determine Residency & Status
|
|
# C# Residency Rule: IsResident iff Zuchtname matches Clan kennel, OR is parent of resident
|
|
resident_guids = set()
|
|
# Step A: Direct Zucht name match
|
|
for a in animals:
|
|
if ex.is_clan_zucht(a.get("zucht")) or ex.is_clan_zucht(a.get("zuchtCanon")):
|
|
resident_guids.add(animal_guid_map[a["id"]])
|
|
|
|
# Step B: Parent of Clan offspring
|
|
# Re-run propagation loop a few times to cover multiple generations
|
|
for _ in range(5):
|
|
for l in litters:
|
|
l_guid = litter_guid_map[l["id"]]
|
|
# Check if any offspring of this litter is resident
|
|
offspring_guids = [animal_guid_map[a["id"]] for a in animals if a.get("litterRef") and a["litterRef"].get("litterId") == l["id"]]
|
|
has_resident_offspring = any(og in resident_guids for og in offspring_guids)
|
|
|
|
if has_resident_offspring:
|
|
p_father, p_mother = litter_parent_map.get(l["id"], [None, None])
|
|
if p_father:
|
|
resident_guids.add(p_father)
|
|
if p_mother:
|
|
resident_guids.add(p_mother)
|
|
|
|
# 6. Construct Gerbils collection
|
|
resolved_gerbils = []
|
|
resolved_photos = []
|
|
|
|
for a in animals:
|
|
a_guid = animal_guid_map[a["id"]]
|
|
|
|
# Resolve ColorVariety
|
|
cv_id = None
|
|
fb_key = a["farbschlag"].strip().lower()
|
|
if fb_key in variety_map:
|
|
cv_id = variety_map[fb_key]
|
|
else:
|
|
# Try matching variants
|
|
for fbv in a.get("farbschlagVariants", []):
|
|
fbv_key = fbv.strip().lower()
|
|
if fbv_key in variety_map:
|
|
cv_id = variety_map[fbv_key]
|
|
break
|
|
|
|
# Resolve Litter
|
|
l_id = None
|
|
l_ref = a.get("litterRef")
|
|
if l_ref and l_ref.get("litterId") and l_ref["litterId"] in litter_guid_map:
|
|
l_id = litter_guid_map[l_ref["litterId"]]
|
|
|
|
# Resolve Contacts
|
|
origin_contact_id = None
|
|
breeder_norm = a.get("breeder", "").strip().lower()
|
|
zucht_norm = a.get("zucht", "").strip().lower()
|
|
if breeder_norm in contact_guid_map:
|
|
origin_contact_id = contact_guid_map[breeder_norm]
|
|
elif zucht_norm in contact_guid_map:
|
|
origin_contact_id = contact_guid_map[zucht_norm]
|
|
|
|
# Dates
|
|
dob = parse_date_only(a.get("dob"))
|
|
death = parse_date_only(a.get("death"))
|
|
|
|
# Gender conversion: M -> male, W -> female
|
|
gender = "Unknown"
|
|
if a.get("gender") == "M":
|
|
gender = "male"
|
|
elif a.get("gender") == "W":
|
|
gender = "female"
|
|
|
|
# Determine Residency
|
|
is_resident = a_guid in resident_guids
|
|
|
|
# Determine Status
|
|
status = "Breeding"
|
|
if death:
|
|
status = "Deceased"
|
|
elif not is_resident:
|
|
status = "GivenAway"
|
|
else:
|
|
# Age presumed deceased (>6 years)
|
|
if dob:
|
|
dt_dob = datetime.strptime(dob, "%Y-%m-%d")
|
|
dt_now = datetime.now()
|
|
age_years = (dt_now - dt_dob).days / 365.25
|
|
if age_years >= 6.0:
|
|
status = "Deceased"
|
|
else:
|
|
status = "Breeding" # default to breeding for resident stock
|
|
|
|
# Raw Import payload
|
|
raw_import_payload = json.dumps({
|
|
"rawGenotype": a["genotype"]["rawGenotype"],
|
|
"unmappedTokens": a["genotype"]["unmappedTokens"],
|
|
"breederText": a.get("breeder", "")
|
|
}, ensure_ascii=False)
|
|
|
|
resolved_gerbils.append({
|
|
"Id": a_guid,
|
|
"Name": a["name"],
|
|
"Gender": gender,
|
|
"Status": status,
|
|
"LitterId": l_id,
|
|
"OriginContactId": origin_contact_id,
|
|
"ReceiverContactId": None,
|
|
"EnclosureId": None,
|
|
"ColorVarietyId": cv_id,
|
|
"DateOfBirth": dob,
|
|
"DateOfDeath": death,
|
|
"CauseOfDeath": None,
|
|
"GoHomeDate": None,
|
|
"Genotype": a["genotype"]["rawGenotype"] if a["genotype"]["rawGenotype"] else None,
|
|
"Notes": None,
|
|
"ImportSource": "docx-export",
|
|
"ExternalRef": a["id"],
|
|
"RawImportData": raw_import_payload,
|
|
"OriginBreeder": a.get("breeder") if a.get("breeder") else (a.get("zucht") if a.get("zucht") else None),
|
|
"NameSearch": ex.norm_name(a["name"]),
|
|
"CharacterTraits": [],
|
|
"CharacterNote": None,
|
|
"IsDeaf": a.get("deaf"),
|
|
"IsResident": is_resident
|
|
})
|
|
|
|
# Attach Photos
|
|
for idx, photo_rel in enumerate(a.get("photos", [])):
|
|
photo_guid = generate_guid(f"photo-{photo_rel}")
|
|
ext = os.path.splitext(photo_rel)[1] or ".jpeg"
|
|
fn_guid = photo_guid.replace("-", "")
|
|
resolved_photos.append({
|
|
"Id": photo_guid,
|
|
"GerbilId": a_guid,
|
|
"FileName": f"{fn_guid}{ext}",
|
|
"SortOrder": idx,
|
|
"_source_path": photo_rel
|
|
})
|
|
|
|
# Set IsBreeder and IsReceiver flags on contacts
|
|
breeder_ids = {g["OriginContactId"] for g in resolved_gerbils if g.get("OriginContactId")}
|
|
receiver_ids = {g["ReceiverContactId"] for g in resolved_gerbils if g.get("ReceiverContactId")}
|
|
for c in resolved_contacts:
|
|
c_id = c["Id"]
|
|
is_breeder = c_id in breeder_ids
|
|
is_receiver = c_id in receiver_ids
|
|
if not is_breeder and not is_receiver:
|
|
is_receiver = True
|
|
c["IsBreeder"] = is_breeder
|
|
c["IsReceiver"] = is_receiver
|
|
|
|
# Save resolved data
|
|
resolved_data = {
|
|
"Contacts": resolved_contacts,
|
|
"Litters": resolved_litters,
|
|
"Gerbils": resolved_gerbils,
|
|
"GerbilPhotos": resolved_photos
|
|
}
|
|
|
|
resolved_path = os.path.join(OUTPUT_DIR, "resolved_import.json")
|
|
with open(resolved_path, "w", encoding="utf-8") as f:
|
|
json.dump(resolved_data, f, ensure_ascii=False, indent=2)
|
|
|
|
# Save resolved CSV files
|
|
save_to_csv(resolved_contacts, os.path.join(OUTPUT_DIR, "resolved_contacts.csv"))
|
|
save_to_csv(resolved_litters, os.path.join(OUTPUT_DIR, "resolved_litters.csv"))
|
|
save_to_csv(resolved_gerbils, os.path.join(OUTPUT_DIR, "resolved_gerbils.csv"))
|
|
save_to_csv(resolved_photos, os.path.join(OUTPUT_DIR, "resolved_photos.csv"))
|
|
|
|
print(f"Success! Saved resolved database-ready records to {resolved_path} (and CSV counterparts):")
|
|
print(f" Contacts: {len(resolved_contacts)}")
|
|
print(f" Litters: {len(resolved_litters)}")
|
|
print(f" Gerbils: {len(resolved_gerbils)}")
|
|
print(f" GerbilPhotos: {len(resolved_photos)}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|