#!/usr/bin/env python3 """ Store-native LLM criteria analysis — runs on the Claude Code CLI (Max subscription), NO external paid API. Reads properties.json, writes the 8-criteria rubric back. Converted off OpenAI/gpt-4o-mini 2026-06-08 → `claude -p` so the whole pipeline runs on the flat-rate subscription. Builds prompt facts from store fields (no HTML fetch — avoids 403 anti-bot blocks). Optional/manual: the default scorer uses the deterministic keyword proxy; run this only when you want LLM-graded criteria on the top candidates. Usage: python3 analyze_store.py # Analyze all active, unscored python3 analyze_store.py --limit 10 # First N (safe first run) python3 analyze_store.py --force # Re-analyze even if criteria exist python3 analyze_store.py --dry-run # Show what would run, no API calls """ import argparse import json import re import subprocess import sys import time from datetime import datetime from pydantic import BaseModel, Field from typing import Literal from store import load, persist, is_active # Schema matches the 8 criteria keys cyber_prairie_score.py reads (verified at # cyber_prairie_score.py:278-291). The legacy PropertyCriteria in # analyze_with_structured_output.py is mis-aligned (has rental_units, missing # food_experience/design_story/livability) — do NOT reuse it. class CPCriteria(BaseModel): market_garden: int = Field(ge=1, le=5, description="Regenerative market garden potential: soil/sun/water/size") guest_accommodation: int = Field(ge=1, le=5, description="B&B / retreat potential: tranquility, setting, existing facilities") workshop: int = Field(ge=1, le=5, description="Workshop / food processing space: outbuildings, room to convert") food_experience: int = Field(ge=1, le=5, description="On-site food experience potential: kitchen, dining space, atmosphere") design_story: int = Field(ge=1, le=5, description="Design / character / story: charm, architectural distinctiveness, narrative pull") location: int = Field(ge=1, le=5, description="Location appeal: accessibility, proximity to interest, visibility") local_market: int = Field(ge=1, le=5, description="Local market access: customer reachability for products/services") livability: int = Field(ge=1, le=5, description="Daily-living quality: utilities, climate, infrastructure, comfort") risk_profile: Literal["Laag", "Gemiddeld", "Hoog"] = Field(description="Overall renovation/viability risk") overall_assessment: str = Field(max_length=300, description="2-3 sentences summarising the property's fit") PROMPT_TEMPLATE = """You evaluate rural properties for a regenerative-farming + B&B project (the user's "Cyber Prairie" goal: a French/Italian rural homestead with food production, guest hosting, and creative workshop). {property_facts} Score each of the 8 criteria on a 1-5 scale: - 1 = very poor / not viable - 2 = poor / significant limitations - 3 = acceptable / some limitations - 4 = good / minor limitations - 5 = excellent / ideal Assess overall risk: - Laag (Low): move-in ready or minimal work - Gemiddeld (Medium): some renovation, moderate uncertainty - Hoog (High): major renovation OR unclear utilities OR questionable viability Be honest. Base scores only on the facts above. Don't speculate beyond evidence.""" def build_facts(p): """Render store-resident facts as the prompt body (replaces HTML fact-extraction).""" parts = ["PROPERTY DATA:"] if p.get('title'): parts.append(f"- Title: {p['title']}") if p.get('location'): parts.append(f"- Location: {p['location']}") if p.get('city'): parts.append(f"- City/Area: {p['city']}") if p.get('search_region'): parts.append(f"- Region: {p['search_region']}") if p.get('country'): parts.append(f"- Country: {p['country']}") if p.get('price'): parts.append(f"- Price: EUR {p['price']:,}") if p.get('building_size') or p.get('building_size_m2'): parts.append(f"- Building size: {p.get('building_size') or p.get('building_size_m2')} m²") if p.get('land_size') or p.get('land_size_m2'): parts.append(f"- Land size: {p.get('land_size') or p.get('land_size_m2')} m²") if p.get('rooms') or p.get('bedrooms'): parts.append(f"- Rooms/bedrooms: {p.get('rooms') or p.get('bedrooms')}") if p.get('renovation_estimate'): parts.append(f"- Renovation needed: {p['renovation_estimate']}") if p.get('keyword_signals'): parts.append(f"- Keywords from listing: {', '.join(p['keyword_signals'])}") if p.get('description'): parts.append(f"\nDESCRIPTION:\n{p['description'][:1500]}") if p.get('gpt_summary'): parts.append(f"\nPRIOR SUMMARY:\n{p['gpt_summary']}") am = p.get('amenities') or {} if am: parts.append("\nAMENITIES:") for k in ('airport', 'bakery', 'hospital', 'train_station', 'supermarket'): v = am.get(k) if isinstance(v, dict) and v.get('name'): parts.append(f"- {k}: {v.get('name')} ({v.get('km', '?')}km)") return "\n".join(parts) def analyze_one(prop): """Single property → criteria dict, via `claude -p` (Claude Code CLI subscription — no external paid API). Raises on failure. JSON is validated through CPCriteria so out-of-range scores / bad enums fail loudly rather than corrupting the store.""" prompt = PROMPT_TEMPLATE.format(property_facts=build_facts(prop)) + ( "\n\nReply with ONLY one JSON object, no prose and no code fences, with exactly " "these keys: market_garden, guest_accommodation, workshop, food_experience, " "design_story, location, local_market, livability (each an integer 1-5), " 'risk_profile (one of "Laag","Gemiddeld","Hoog"), ' "overall_assessment (a 2-3 sentence string, max 280 characters)." ) t0 = time.time() proc = subprocess.run(['claude', '-p', prompt], capture_output=True, text=True, timeout=180) out = (proc.stdout or '').strip() m = re.search(r'\{.*\}', out, re.S) if not m: raise ValueError(f'no JSON from claude -p: {out[:80]!r}') raw = json.loads(m.group(0)) if isinstance(raw.get('overall_assessment'), str): raw['overall_assessment'] = raw['overall_assessment'][:300] # clamp to schema cap c = CPCriteria(**raw) # validates ranges + enum # Note: risk_profile NOT authoritative — cyber_prairie_score.py warns the LLM text # label is unreliable; the numeric risk_score from georisques wins. Kept for reference. return { 'criteria': { 'market_garden': c.market_garden, 'guest_accommodation': c.guest_accommodation, 'workshop': c.workshop, 'food_experience': c.food_experience, 'design_story': c.design_story, 'location': c.location, 'local_market': c.local_market, 'livability': c.livability, }, 'gpt_summary': c.overall_assessment, 'gpt_risk_profile_unverified': c.risk_profile, # legacy ref, not used by scorer 'gpt_analyzed_at': datetime.now().isoformat(), 'gpt_model': 'claude-cli', '_cost': 0.0, # flat-rate subscription — no metered cost '_duration': time.time() - t0, } def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument('--limit', type=int, default=0, help='Max properties to analyze (0=all)') ap.add_argument('--force', action='store_true', help='Re-analyze even if criteria exist') ap.add_argument('--dry-run', action='store_true', help='Print plan, no API calls') args = ap.parse_args() store = load() pending = [] REQUIRED_KEYS = {'food_experience', 'design_story', 'livability', 'market_garden', 'guest_accommodation', 'workshop', 'location', 'local_market'} for url, p in store.items(): if not is_active(p): continue # Need minimum data to analyze if not (p.get('title') or p.get('city')) or not p.get('price'): continue crit = p.get('criteria') or {} if not args.force: # Skip only if criteria has the full 8-key shape the scorer expects. # Truncated 6-key records (from earlier broken schema) still go through. if crit and REQUIRED_KEYS.issubset(crit.keys()): continue pending.append((url, p)) if args.limit: pending = pending[:args.limit] print(f" Pending analysis: {len(pending)} properties") if args.dry_run or not pending: for url, p in pending[:5]: print(f" - {p.get('search_region','?'):15} €{p.get('price','?'):>8} {(p.get('title') or p.get('city',''))[:60]}") if len(pending) > 5: print(f" ... and {len(pending)-5} more") return 0 total_cost = 0.0 ok = 0 fail_reasons = {} # keep cardinality of distinct failure modes for i, (url, p) in enumerate(pending, 1): try: result = analyze_one(p) except Exception as e: reason = f'{type(e).__name__}: {str(e)[:60]}' fail_reasons[reason] = fail_reasons.get(reason, 0) + 1 print(f" [{i}/{len(pending)}] FAIL {url[:60]}: {reason}") continue cost = result.pop('_cost'); dur = result.pop('_duration') store[url].update(result) total_cost += cost ok += 1 if i % 10 == 0 or i == len(pending): persist(store) print(f" [{i}/{len(pending)}] saved ok={ok} cost=${total_cost:.4f}") persist(store) n_fail = sum(fail_reasons.values()) print(f"\n Done: {ok}/{len(pending)} analyzed, total cost ${total_cost:.4f} (avg ${total_cost/max(ok,1):.5f}/prop)") if fail_reasons: print(f" Failures: {n_fail} total — breakdown:") for reason, count in sorted(fail_reasons.items(), key=lambda x: -x[1]): print(f" {count}x {reason}") # Fail loud if >50% failed — pipeline.py reads exit code if pending and n_fail > len(pending) / 2: print(f" WARN: majority failed — returning non-zero exit code") return 1 return 0 if __name__ == '__main__': sys.exit(main())