#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Site Hotspot Intelligent Scoring System v2.0 ============================================ Auto charging station site evaluation: 1. Auto geocode (address -> lat/lng) - Tencent Maps Geocoder 2. Amap nearby charging stations (count/guns/fee/rate) 3. Tencent Maps POI cross-verification 4. AI geography analysis (traffic/heat/flow/EV ranking) 5. Four-dimension scoring + competition penalty (each >10gun station = -20pts) 6. Cost accounting and ROI calculation 7. Final decision: <55 no-go / ~58-62 discuss / >=65 go Usage: python site_scorer.py "Address" # Single site scoring python site_scorer.py "Address" --upload # Score + auto-upload to CRM python site_scorer.py --batch sites.csv # Batch scoring python site_scorer.py --batch sites.csv --upload # Batch score + upload all python site_scorer.py --lat 25.5 --lng 119.8 # Direct coordinates python site_scorer.py --dry-run # Show rules only, no API calls """ import urllib.request, urllib.parse, json, csv, time, sys, os, re, ssl, math, datetime, collections, traceback from html import escape as _e # ======================== CONFIG ======================== TENCENT_KEY = "LHOBZ-KKNWT-YMOXB-VSMJC-VJJJQ-BBB6P" AMAP_KEY = "8ea4f0ac00dfb092d3e14d5776b1def9" TENCENT_SEARCH_KEY = TENCENT_KEY CRM_BASE = "https://crm.ixgn.cn" SCORE_NOGO = 55 SCORE_DISCUSS_LO = 58 SCORE_DISCUSS_HI = 62 SCORE_GO = 65 COMPETE_PENALTY = 20 RADIUS_KM = 2 DEFAULT_COST = { "equip_w": 100, "install_w": 30, "rent_wy": 14, "opex_wy": 8, "elec_cost": 0.65, "n_guns": 10, "kw_per_gun": 120, "util_pct": 18, } BENCH_FAST_FEE = 0.30 BENCH_HEAVY_FEE = 0.20 STORE_FILE = "site_score_report.json" CTX = ssl.create_default_context() # ======================== UTILITIES ======================== def _get(url, timeout=15): req = urllib.request.Request(url) try: with urllib.request.urlopen(req, context=CTX, timeout=timeout) as r: return json.loads(r.read().decode()) except Exception as e: return {"error": str(e)} def norm(s): return re.sub(r"\s+", "", str(s or "")).strip() def now_str(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M") # ======================== 1. GEOCODING (Tencent) ======================== def geocode(address): if not address: return None url = ("https://apis.map.qq.com/ws/geocoder/v1/?address=" + urllib.parse.quote(address) + "&key=" + TENCENT_KEY) d = _get(url) if d.get("status") == 0: loc = d.get("result", {}).get("location", {}) ac = d.get("result", {}).get("address_component", {}) return { "lng": loc.get("lng"), "lat": loc.get("lat"), "address": d.get("result", {}).get("address"), "city": ac.get("city", ""), "district": ac.get("district", ""), "province": ac.get("province", ""), "formatted": d.get("result", {}).get("title", address), "source": "tencent", } print(" [Tencent geocode failed] " + d.get("message", str(d)) + " -> trying Amap fallback") am = amap_geocode(address) if am: return am return None def amap_geocode(address): if not AMAP_KEY: return None url = ("https://restapi.amap.com/v3/geocode/geo?key=" + AMAP_KEY + "&address=" + urllib.parse.quote(address)) d = _get(url, timeout=15) if d.get("status") != "1" or not d.get("geocodes"): return None g = d["geocodes"][0] loc = (g.get("location") or "").split(",") if len(loc) != 2: return None return { "lng": float(loc[0]), "lat": float(loc[1]), "address": g.get("formatted_address", address), "city": g.get("city", "") or g.get("province", ""), "district": g.get("district", ""), "province": g.get("province", ""), "formatted": g.get("formatted_address", address), "source": "amap", } def reverse_geocode(lat, lng): url = f"https://apis.map.qq.com/ws/geocoder/v1/?location={lat},{lng}&key={TENCENT_KEY}" d = _get(url) if d.get("status") != 0: return {"formatted": f"{lat},{lng}", "city": "", "district": ""} r = d.get("result", {}) ac = r.get("address_component", {}) return { "formatted": r.get("address", f"{lat},{lng}"), "city": ac.get("city", ""), "district": ac.get("district", ""), "province": ac.get("province", ""), } # ======================== 2. AMAP NEARBY STATIONS ======================== def amap_nearby_stations(lat, lng, radius=2000, keywords="充电桩"): if not AMAP_KEY: return {"stations": [], "summary": {"total": 0, "note": "No Amap key configured"}} url = ("https://restapi.amap.com/v3/place/around?" f"key={AMAP_KEY}&location={lng},{lat}&keywords=" + urllib.parse.quote(keywords) + f"&radius={radius}&offset=50&output=json&extensions=all") d = _get(url, timeout=20) if "error" in d or d.get("status") != "1": return {"stations": [], "summary": {"total": 0, "error": d.get("info", str(d))}} pois = d.get("pois", []) stations = [] big_stations = 0 for p in pois: name = p.get("name", "") guns = extract_guns(name) dist = float(p.get("distance", 0)) fee_info = parse_amap_fee(p) is_big = guns >= 10 if is_big: big_stations += 1 stations.append({ "name": name, "address": p.get("address", ""), "type": p.get("type", ""), "dist_m": int(dist), "guns": guns, "is_big": is_big, "location": p.get("location", "").split(","), "fee_service": fee_info.get("service_fee"), "fee_elec": fee_info.get("elec_fee"), "rating": p.get("rating", ""), }) return { "stations": sorted(stations, key=lambda x: x["dist_m"]), "summary": {"total": len(stations), "big_stations": big_stations, "radius_km": radius // 1000}, } def extract_guns(name): # Large-station proxies: super-/heavy-truck/energy-storage charging hubs -> >=10 guns if any(k in name for k in ["超充", "重卡", "光储", "储充", "大功率", "mega", " Megacharge"]): return 12 m = re.search(r'(\d+)\s*[guns|piles]', name, re.I) if m: return int(m.group(1)) m = re.search(r'(\d+)\s*guns', name) if m: return int(m.group(1)) if any(k in name for k in ["super charge", "high power", "fast charge"]): return 8 if any(k in name for k in ["station", "pile"]): return 4 return 4 def parse_amap_fee(poi): cost_str = poi.get("cost", "") ext = poi.get("ext_info", {}) or {} fee_elec = None fee_svc = None if cost_str: m = re.search(r'(\d+\.?\d*)\s*yuan', cost_str) if m: fee_svc = float(m.group(1)) if isinstance(ext, dict): efee = ext.get("charge_fee") if efee: try: fee_elec = float(efee) except: pass return {"service_fee": fee_svc, "elec_fee": fee_elec} # ======================== 3. TENCENT POI VERIFICATION ======================== def tencent_nearby_poi(lat, lng, keyword, radius=2000): url = ("https://apis.map.qq.com/ws/place/v1/search?" f"boundary=nearby({lat},{lng},{radius},0)&keyword=" + urllib.parse.quote(keyword) + f"&page_size=25&page_index=1&key={TENCENT_KEY}") d = _get(url, timeout=12) if "error" in d or d.get("status") != 0: return None return d.get("data", []) TENCENT_DOWN = False def tencent_verify_competition(lat, lng): global TENCENT_DOWN if TENCENT_DOWN: return {"_skipped": "tencent unavailable (quota/key)"} results = {} errs = 0 for kw in ["充电桩", "充电站", "超充站"]: pois = tencent_nearby_poi(lat, lng, kw) if pois is None: errs += 1; results[kw] = 0 else: results[kw] = len(pois) time.sleep(0.12) if errs == 3: TENCENT_DOWN = True # exhausted key detected -> skip for rest of run return results # ======================== 4. FOUR-DIMENSION SCORING ENGINE ======================== class SiteScorer: """ Four-dimension scoring system: D1 Station count (>10gun within 2km) - fewer is better D2 Nearby preferred station charge rate (historical/AI estimate) - higher means more demand D3 Service fee (vs benchmark) - near benchmark is best D4 Geography quality (traffic/heat/flow/city EV ranking) - AI comprehensive score """ def __init__(self, site_data, amap_data=None, tencent_verify=None, ai_geo=None): self.site = site_data self.amap = amap_data or {"stations": [], "summary": {}} self.tencent = tencent_verify or {} self.ai_geo = ai_geo or {} self.details = {} self.dim_scores = {} self.penalties = [] self.bonuses = [] def score(self): d1 = self._score_d1() d2 = self._score_d2() d3 = self._score_d3() d4 = self._score_d4() self.dim_scores = {"D1_StationCount": d1, "D2_ChargeRate": d2, "D3_ServiceFee": d3, "D4_Geography": d4} raw_total = sum(self.dim_scores.values()) / len(self.dim_scores) # Competition penalty big_n = self.amap.get("summary", {}).get("big_stations", 0) if big_n > 0: penalty = min(big_n * COMPETE_PENALTY, 60) self.penalties.append(f"Nearby {big_n} large competitor stations(>=10guns), each -{COMPETE_PENALTY}pts = -{penalty}pts") raw_total -= penalty bonus_total = sum(b["score"] for b in self.bonuses) final_score = max(0, min(100, raw_total + bonus_total)) decision, level, advice = self._decision(final_score) dims_met = sum(1 for v in self.dim_scores.values() if v >= 60) return { "total": round(final_score, 1), "raw_total": round(raw_total, 1), "dim_scores": {k: round(v, 1) for k, v in self.dim_scores.items()}, "dims_met": dims_met, "dims_total": 4, "penalties": self.penalties, "bonuses": self.bonuses, "decision": decision, "level": level, "advice": advice, "details": self.details, } def _score_d1(self): big_n = self.amap.get("summary", {}).get("big_stations", 0) total_n = self.amap.get("summary", {}).get("total", 0) stations = self.amap.get("stations", []) if big_n == 0: score = 100; detail = "No large competitor stations(>=10guns) within 2km, blue ocean" elif big_n == 1: score = 70; detail = f"1 large competitor station within 2km, manageable competition" elif big_n == 2: score = 45; detail = f"2 large competitor stations within 2km, intense competition" else: score = 20; detail = f"{big_n} large competitor stations within 2km, red ocean area" if total_n > 10: score = max(score - 10, 10); detail += " (total " + str(total_n) + " stations, high density)" self.details["D1"] = { "score": score, "big_stations": big_n, "total_stations": total_n, "detail": detail, "nearby_top5": [ {"name": s["name"], "dist_m": s["dist_m"], "guns": s["guns"]} for s in stations[:5] ] } return score def _score_d2(self): stations = self.amap.get("stations", []) if not stations: score = 50; detail = "No nearby station data, neutral estimate" ai_demand = self.ai_geo.get("demand_level") if ai_demand == "high": score = 65; detail = "No station data; AI judges regional demand HIGH" elif ai_demand == "low": score = 40; detail = "No station data; AI judges regional demand LOW" else: rated = [s for s in stations if s.get("rating")] avg_rating = 0 if rated: avg_rating = sum(float(s["rating"]) for s in rated) / len(rated) if avg_rating >= 4: score = 75; detail = f"Avg nearby station rating {avg_rating:.1f}, good user feedback, demand confirmed" elif avg_rating >= 3: score = 60; detail = f"Avg nearby station rating {avg_rating:.1f}, average demand" else: score = 45; detail = "Low nearby station ratings or no data" ai_demand = self.ai_geo.get("demand_level") if ai_demand == "high": score = min(score + 15, 95); detail += " | AI judges regional demand as HIGH" elif ai_demand == "low": score = max(score - 15, 20); detail += " | AI judges regional demand as LOW" self.details["D2"] = {"score": score, "detail": detail} return score def _score_d3(self): stations = self.amap.get("stations", []) if not stations: score = 60; detail = "No nearby service fee data, neutral industry average" self.details["D3"] = {"score": score, "detail": detail, "bench_fast": BENCH_FAST_FEE, "bench_heavy": BENCH_HEAVY_FEE} return score known_fees = [s["fee_service"] for s in stations if s.get("fee_service")] if known_fees: avg_fee = sum(known_fees) / len(known_fees) diff = avg_fee - BENCH_FAST_FEE if abs(diff) <= 0.05: score = 85; detail = f"Avg nearby service fee {avg_fee:.2f}, near fast-charge benchmark {BENCH_FAST_FEE}, healthy pricing" elif diff > 0.05: score = max(70 - (diff - 0.05) * 100, 30); detail = f"Avg nearby fee {avg_fee:.2f}, above benchmark by {diff:.2f}, may suppress demand" else: score = max(50 + diff * 80, 15); detail = f"Avg nearby fee {avg_fee:.2f}, below benchmark by {abs(diff):.2f}, possible price war zone" else: score = 55; detail = "No public fee data for nearby stations, field research needed" ai_fee = self.ai_geo.get("fee_environment") if ai_fee == "healthy": score = min(score + 10, 95) elif ai_fee == "price_war": score = max(score - 20, 15) self.details["D3"] = { "score": score, "detail": detail, "bench_fast": BENCH_FAST_FEE, "bench_heavy": BENCH_HEAVY_FEE, "avg_observed_fee": known_fees[0] if len(known_fees) == 1 else (sum(known_fees)/len(known_fees) if known_fees else None) } return score def _score_d4(self): ai = self.ai_geo if not ai: score = 50; detail = "No AI geography data" self.details["D4"] = {"score": score, "detail": detail} return score scores = [] reasons = [] traffic = ai.get("traffic_convenience", 3) s = traffic * 20; scores.append(s) reasons.append(f"Traffic convenience:{traffic}/5 -> {s:.0f}pts") heat = ai.get("heat_index", 3) s = heat * 20; scores.append(s) reasons.append(f"Regional heat:{heat}/5 -> {s:.0f}pts") flow = ai.get("traffic_flow", 3) s = flow * 18; scores.append(s) reasons.append(f"Traffic flow:{flow}/5 -> {s:.0f}pts") ev_rank = ai.get("city_ev_ranking", 50) s = ev_rank; scores.append(s) reasons.append(f"City EV ownership rank percentile:{ev_rank:.0f}% -> {s:.0f}pts") pref = ai.get("preference_score", 60) s = pref * 0.22; scores.append(s) reasons.append(f"AI preference score:{pref:.0f}/100 -> {s:.0f}pts") score = sum(scores) / len(scores) if scores else 50 detail = " | ".join(reasons) if ai.get("fixed_fleet"): score = min(score + 12, 98); reasons.append("Fixed fleet guarantee +12pts") if ai.get("has_support_facilities"): score = min(score + 8, 98); reasons.append("Support facilities +8pts") self.details["D4"] = {"score": round(score, 1), "detail": detail, "sub_items": reasons} return score def _decision(self, score): dims_met = sum(1 for v in self.dim_scores.values() if v >= 60) if score < SCORE_NOGO: return "NOT RECOMMENDED", "NOGO", ( f"Score {score:.1f} < no-go line ({SCORE_NOGO}), too risky. " f"Suggest abandon or renegotiate terms. {dims_met}/4 dimensions met.") elif SCORE_DISCUSS_LO <= score <= SCORE_DISCUSS_HI: return "NEEDS DISCUSSION", "DISCUSS", ( f"Score {score:.1f} in discussion range [{SCORE_DISCUSS_LO},{SCORE_DISCUSS_HI}], " f"opportunity but cautious. Suggest small meeting to discuss: reduce rent/revenue share/scale down. " f"{dims_met}/4 dimensions met.") elif score >= SCORE_GO: if dims_met >= 3: return "STRONGLY RECOMMENDED", "GO_STRONG", ( f"Score {score:.1f} >= go-line ({SCORE_GO}), {dims_met}/4 dimensions met, " f"advantageous site. Recommend proceeding, prioritize development resources.") else: return "RECOMMENDED", "GO", ( f"Score {score:.1f} >= go-line ({SCORE_GO}), but only {dims_met}/4 dimensions met. " f"Doable, focus on strengthening weak dimensions.") else: return "EDGE CASE", "EDGE", ( f"Score {score:.1f} in edge range, {dims_met}/4 dimensions met. " f"Can proceed with good terms, suggest negotiating better conditions.") # ======================== 5. ROI CALCULATION ======================== def calc_roi(site_name, scorer_result, cost_overrides=None): c = dict(DEFAULT_COST) if cost_overrides: c.update(cost_overrides) guns = c["n_guns"]; kw = c["kw_per_gun"]; util = c["util_pct"] / 100 fee = scorer_result.get("details", {}).get("D3", {}).get("avg_observed_fee") or 0.35 elec_cost = c["elec_cost"] max_kwh_day = guns * kw * 24 actual_kwh_day = max_kwh_day * util # Revenue = electricity pass-through at cost + service fee (pure margin); # electricity purchase cost offsets the pass-through, so gross = service-fee income. annual_rev = actual_kwh_day * (elec_cost + fee) * 365 / 10000 annual_elec = actual_kwh_day * elec_cost * 365 / 10000 gross = annual_rev - annual_elec annual_fixed = c["rent_wy"] + c["opex_wy"] net = gross - annual_fixed cash_invest = c["install_w"] + c["equip_w"] payback_years = cash_invest / net if net > 0 else 999 be_util = (annual_fixed * 10000) / ((fee - elec_cost) * max_kwh_day * 365) * 100 npv = -cash_invest for y in range(1, 6): npv += net / (1.08 ** y) score = scorer_result.get("total", 50) if score >= 70: util_adj = "High-score site, utilization may exceed expectations" elif score >= 60: util_adj = "Medium-high, needs operational push" else: util_adj = "Low-score site, utilization risk is high" return { "invest_cash_w": round(cash_invest, 1), "annual_rev_w": round(annual_rev, 1), "annual_elec_w": round(annual_elec, 1), "gross_w": round(gross, 1), "annual_fixed_w": round(annual_fixed, 1), "net_annual_w": round(net, 1), "payback_years": round(payback_years, 1), "be_util_pct": round(be_util, 1), "npv_5y_w": round(npv, 1), "irr_estimate": round((net / cash_invest) * 100, 1) if cash_invest > 0 else 0, "util_adjustment": util_adj, "params_used": c, "assumed_fee": fee, "actual_kwh_day": round(actual_kwh_day, 0), } # ======================== 6. AI GEOGRAPHY ANALYSIS ======================== def analyze_geography_ai(address, lat, lng, city): result = { "traffic_convenience": 3, "heat_index": 3, "traffic_flow": 3, "city_ev_ranking": 50, "preference_score": 60, "demand_level": "medium", "fee_environment": "unknown", "estimated_utilization": 0.18, "fixed_fleet": False, "has_support_facilities": False, "analysis_source": "rule-based estimation", } city_cn_rank = { "福州": 77, "厦门": 84, "泉州": 70, "漳州": 54, "莆田": 48, "宁德": 45, "龙岩": 42, "三明": 38, "南平": 35, "平潭": 30, } matched = False for cn, v in city_cn_rank.items(): if cn in (address or "") or cn in (city or ""): result["city_ev_ranking"] = v; matched = True; break if not matched: en_rank = {"Fuzhou": 77, "Xiamen": 84, "Quanzhou": 70, "Zhangzhou": 54, "Putian": 48, "Ningde": 45, "Longyan": 42, "Sanming": 38, "Nanping": 35, "Pingtan": 30} cl = norm(city).lower() for en, v in en_rank.items(): if en.lower() in cl: result["city_ev_ranking"] = v; matched = True; break full_text = norm(str(address) + " " + str(city)) high_traffic_kw = ["expressway","national highway","hub","airport","railway station", "logistics park","industrial park","development zone","port","dock"] med_traffic_kw = ["main road","avenue","center","plaza","mall","commercial"] hot_kw = ["cbd","center","commercial","plaza","pedestrian street","wanda","baolong","taihe"] cold_kw = ["remote","suburb","rural","village","mountain","ridge"] if any(k in full_text for k in high_traffic_kw): result["traffic_convenience"] = 4.5; result["traffic_flow"] = 4.2 elif any(k in full_text for k in med_traffic_kw): result["traffic_convenience"] = 3.5; result["traffic_flow"] = 3.2 if any(k in full_text for k in hot_kw): result["heat_index"] = 4.2 elif any(k in full_text for k in cold_kw): result["heat_index"] = 2.2 industry_kw = { "logistics": ("fixed_fleet", True, "logistics fleet guarantee"), "freight": ("fixed_fleet", True, "freight vehicle guarantee"), "ride-hailing": ("fixed_fleet", True, "ride-hailing hub"), "courier": ("fixed_fleet", True, "courier distribution guarantee"), "park": ("has_support_facilities", True, "industrial park facilities"), "warehouse": ("has_support_facilities", True, "warehouse support"), } for k, (field, val, note) in industry_kw.items(): if k in full_text: result[field] = val; break subs = [result["traffic_convenience"], result["heat_index"], result["traffic_flow"], result["city_ev_ranking"] / 20] result["preference_score"] = sum(subs) / len(subs) * 20 if result["preference_score"] >= 70: result["demand_level"] = "high" elif result["preference_score"] >= 50: result["demand_level"] = "medium" else: result["demand_level"] = "low" return result # ======================== 7. MAIN FLOW ======================== def evaluate_site(address=None, lat=None, lng=None, name=None, cost_overrides=None, dry_run=False, skip_tencent=False): report = { "timestamp": now_str(), "input": {"address": address, "lat": lat, "lng": lng, "name": name}, "geocode": None, "amap": None, "tencent_verify": None, "ai_geo": None, "scorer": None, "roi": None, } print("\n[1/6] Geocoding...") if lat and lng: geo = reverse_geocode(lat, lng) geo["lat"] = lat; geo["lng"] = lng elif address: geo = geocode(address) else: return {**report, "error": "Must provide address or coordinates"} if not geo: return {**report, "error": "Geocoding failed (Tencent daily quota exhausted and no Amap key). Use --lat/--lng to bypass, or set AMAP_KEY, or retry tomorrow after quota reset."} report["geocode"] = geo print(f" OK: {geo.get('formatted')} ({geo['lat']}, {geo['lng']})") if dry_run: print(" [dry-run] Skipping API queries") return report clat, clng = geo["lat"], geo["lng"] print("[2/6] Amap nearby charging stations...") amap_data = amap_nearby_stations(clat, clng, RADIUS_KM * 1000) report["amap"] = amap_data smry = amap_data["summary"] print(f" OK: {smry['total']} stations (large ones >=10guns: {smry.get('big_stations', 0)})") print("[3/6] Tencent Maps POI verification...") tc_v = {} if skip_tencent else tencent_verify_competition(clat, clng) report["tencent_verify"] = tc_v print(f" OK: {tc_v}") print("[4/6] AI geography analysis...") ai_geo = analyze_geography_ai( (address or name or geo.get("formatted", "")), clat, clng, geo.get("city", "")) report["ai_geo"] = ai_geo print(f" OK: preference {ai_geo['preference_score']:.0f}/100, demand {ai_geo['demand_level']}") print("[5/6] Four-dimension scoring...") scorer = SiteScorer(geo, amap_data, tc_v, ai_geo) score_result = scorer.score() report["scorer"] = score_result print(f" OK: Total {score_result['total']} -> {score_result['decision']}") for dim, sc in score_result["dim_scores"].items(): print(f" {dim}: {sc}") print("[6/6] Cost & ROI calculation...") roi = calc_roi(name or address or "Unnamed site", score_result, cost_overrides) report["roi"] = roi print(f" OK: Annual net {roi['net_annual_w']}wan | Payback {roi['payback_years']}y | NPV5Y={roi['npv_5y_w']}wan") return report # ======================== 8. REPORT OUTPUT ======================== def generate_report_html(report): g = report.get("geocode") or {} sc = report.get("scorer") or {} roi = report.get("roi") or {} amap = report.get("amap") or {} ai = report.get("ai_geo") or {} details = sc.get("details", {}) ts = report.get("timestamp", "") level = sc.get("level", "") if level in ("NOGO",): vc, vb = "#dc2626", "linear-gradient(135deg,#dc2626,#ef4444)" elif level in ("DISCUSS", "EDGE"): vc, vb = "#ea580c", "linear-gradient(135deg,#ea580c,#f59e0b)" elif level == "GO_STRONG": vc, vb = "#16a34a", "linear-gradient(135deg,#16a34a,#22c55e)" else: vc, vb = "#2563eb", "linear-gradient(135deg,#2563eb,#3b82f6)" dim_icons = {"D1_StationCount": "\U0001f3ed", "D2_ChargeRate": "\u26a1", "D3_ServiceFee": "\U0001f4b0", "D4_Geography": "\U0001f5fa"} dim_cards = "" for dim_key, dim_val in sc.get("dim_scores", {}).items(): icon = dim_icons.get(dim_key, "\U0001f4ca") d_short = dim_key.split("_")[1] if "_" in dim_key else dim_key d_detail = details.get(d_short[:2], {}) if len(d_short) >= 2 else details.get(dim_key, {}) d_detail_text = d_detail.get("detail", "") if isinstance(d_detail, dict) else str(d_detail) color = "#16a34a" if dim_val >= 70 else ("#ea580c" if dim_val >= 50 else "#dc2626") dim_cards += ('
' '
' + icon + '
' '
' + d_short + '
' '
' + f'{dim_val:.0f}' + '/100
' '
' + _e(d_detail_text) + '
') d1 = details.get("D1", {}) top5 = d1.get("nearby_top5", []) station_rows = "" for s in top5: tag = '>=10guns' if s.get("is_big") else "normal" station_rows += ("" + _e(s["name"]) + "" + str(s["dist_m"]) + "m" + str(s["guns"]) + "guns" + tag + "") penalty_html = "" for p in sc.get("penalties", []): penalty_html += '
\u26a0\ufe0f ' + _e(p) + '
' for b in sc.get("bonuses", []): penalty_html += '
\u2705 ' + _e(b.get("text", "")) + '
' roi_rows = [ ("\U0001f4b3 Your cash investment (wan)", str(roi.get("invest_cash_w", "-")) + " wan", "Install + equipment one-time"), ("\U0001f4c8 Annual revenue (wan)", str(roi.get("annual_rev_w", "-")) + " wan", "Daily " + str(roi.get("actual_kwh_day", "-")) + "kWh x fee " + str(roi.get("assumed_fee", "-"))), ("\u26a1 Annual electricity cost (wan)", str(roi.get("annual_elec_w", "-")) + " wan", "Purchase cost " + str(roi.get("params_used", {}).get("elec_cost", "-")) + "/kWh"), ("\U0001f4ca Gross profit (wan)", str(roi.get("gross_w", "-")) + " wan", "Revenue - electricity"), ("\U0001f3e2 Annual fixed cost (wan)", str(roi.get("annual_fixed_w", "-")) + " wan", "Rent " + str(roi.get("params_used", {}).get("rent_wy", "-")) + " + O&M " + str(roi.get("params_used", {}).get("opex_wy", "-"))), ("\U0001f48e Net annual cash flow (wan)", str(roi.get("net_annual_w", "-")) + " wan", "Gross - fixed costs"), ("\u23f1 Payback period (years)", str(roi.get("payback_years", "-")) + " y", "Cash investment / annual net"), ("\U0001f4c9 Break-even utilization", str(roi.get("be_util_pct", "-")) + "%", "Below this = loss"), ("\U0001f4e6 5-year NPV @8% discount (wan)", str(roi.get("npv_5y_w", "-")) + " wan", "Est. IRR ~" + str(roi.get("irr_estimate", "-")) + "%"), ] roi_table = "".join( '' + r[0] + '' + r[1] + '' + r[2] + '' for r in roi_rows ) geo_items = "" d4 = details.get("D4", {}) if isinstance(d4, dict): for sub in d4.get("sub_items", []): geo_items += '
' + _e(sub) + '
' tc_html = "" for k, v in report.get("tencent_verify", {}).items(): tc_html += '' + _e(k) + ': ' + str(v) + ' ' html = ('\n' '' 'Site Hotspot Intelligent Scoring Report\n
' '

\U0001f507 Site Hotspot Intelligent Scoring Report

' '
Four-dim auto-scoring | Amap/Tencent dual-source | AI geography | Cost ROI
' '
\U0001f4cd ' + _e(g.get("formatted", report.get("input", {}).get("address", ""))) + '
' + ts + " | City: " + _e(g.get("city", "")) + " | Coords: " + str(g.get("lat", "-")) + "," + str(g.get("lng", "-")) + "
" '
' + str(sc.get("total", "-")) + '
Total Score / 100
' '
' + _e(sc.get("advice", "")) + '
' "Dimensions met: " + str(sc.get("dims_met", 0)) + "/" + str(sc.get("dims_total", 4)) + " | Raw total: " + str(sc.get("raw_total", "-")) + "
" '

\U0001f4ca Four-dimension Scoring Details

' '
' + dim_cards + "
" + ('

\u2696\ufe0f Adjustments

' + penalty_html + "
" if penalty_html else "") + '

\U0001f3ed Nearby 2km Charging Stations (Amap)

' '

Total: ' + str(amap.get("summary", {}).get("total", 0)) + " stations, large(>=10guns): " + str(amap.get("summary", {}).get("big_stations", 0)) + "

" '' + (station_rows or '') + "
NameDistanceGunsType
No data
" + ('

\U0001f5fa Tencent Maps Cross-verification

' + tc_html + "
" if tc_html else "") + '

\U0001f916 AI Geography Analysis

' '
' '
' '
' + f'{ai.get("preference_score", 0):.0f}' + '
AI Preference Score
' '
' '
' + f'{ai.get("city_ev_ranking", 0):.0f}' + '%
City EV Ranking Percentile
' + geo_items + '
Source: ' + _e(ai.get("analysis_source", "-")) + "
" '

\U0001f4b0 Cost Accounting & ROI

' '' + roi_table + "
MetricValueNote
" '

\U0001f4db Scoring Rules Reference

" '
Site Hotspot Scoring System v2.0 | Zhiwei Xiaoyuan CRM suite
' "Data: Amap | Tencent Maps | AI Geography Engine | Reference only, not investment advice
" "") return html # ======================== 9. CRM AUTO-UPLOAD (crm.ixgn.cn) ======================== CRM_USER, CRM_PW = "admin", "88073100" _crm_token = None SITE_UPLOAD_STORE = "site_hotspot_upload_store.json" def crm_login(): global _crm_token req = urllib.request.Request(CRM_BASE + "/api/auth/login", data=json.dumps({"username": CRM_USER, "password": CRM_PW}).encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, context=CTX, timeout=15) as r: _crm_token = json.load(r).get("token") return _crm_token def _crm_hdr(): global _crm_token if not _crm_token: crm_login() return {"Authorization": "Bearer " + _crm_token, "Content-Type": "application/json"} def crm_post(path, body): global _crm_token for _ in range(3): try: req = urllib.request.Request(CRM_BASE + path, data=json.dumps(body).encode(), headers=_crm_hdr(), method="POST") with urllib.request.urlopen(req, context=CTX, timeout=30) as r: return json.load(r) except urllib.error.HTTPError as e: if e.code == 401: crm_login(); continue raise raise Exception("CRM POST failed " + path) def crm_get(path): global _crm_token for _ in range(3): try: req = urllib.request.Request(CRM_BASE + path, headers=_crm_hdr()) with urllib.request.urlopen(req, context=CTX, timeout=25) as r: return json.load(r) except urllib.error.HTTPError as e: if e.code == 401: crm_login(); continue raise raise Exception("CRM GET failed " + path) def _level_label(level): return {"GO_STRONG": "推荐", "GO": "可做", "DISCUSS": "商讨", "EDGE": "边缘", "NOGO": "不推荐"}.get(level or "", level or "") _LEVEL_PRI = {"GO_STRONG": "高", "GO": "高", "DISCUSS": "中", "EDGE": "中", "NOGO": "低"} def site_upload_key(name, region): return norm(name) + "|" + norm(region) def load_site_store(): if os.path.exists(SITE_UPLOAD_STORE): try: return json.load(open(SITE_UPLOAD_STORE, encoding="utf-8")) except: pass return {} def save_site_store(s): json.dump(s, open(SITE_UPLOAD_STORE, "w", encoding="utf-8"), ensure_ascii=False, indent=2) def upload_report_to_crm(report): """Upload one scored site to CRM /api/target-sites. Returns result dict.""" g = report.get("geocode") or {} sc = report.get("scorer") or {} roi = report.get("roi") or {} ai = report.get("ai_geo") or {} ds = sc.get("dim_scores", {}) inp = report.get("input", {}) name = inp.get("name") or g.get("formatted") or inp.get("address") or "未命名场站" addr = g.get("formatted") or g.get("address") or inp.get("address") or "" district = g.get("district", "") region = (g.get("city", "") + ("·" + district if district else "")) or "未定位" total = sc.get("total") total_s = f"{total:.0f}" if isinstance(total, (int, float)) else str(total or "-") level = sc.get("level", "") priority = _LEVEL_PRI.get(level, "中") note = ( f"【热点评分·{total_s}分·{_level_label(level)}】\n" f"建议: {sc.get('advice','')}\n" f"四维: 场站数D1={ds.get('D1_StationCount','')} 充电率D2={ds.get('D2_ChargeRate','')} " f"服务费D3={ds.get('D3_ServiceFee','')} 地理D4={ds.get('D4_Geography','')}\n" f"满足维度: {sc.get('dims_met',0)}/{sc.get('dims_total',4)}\n" f"ROI: 年净{roi.get('net_annual_w','')}万 回收{roi.get('payback_years','')}年 " f"NPV5年{roi.get('npv_5y_w','')}万\n" f"城市EV排名:{ai.get('city_ev_ranking','')} 偏好:{ai.get('preference_score','')}\n" f"坐标:{g.get('lat','')},{g.get('lng','')}\n" f"来源:场站热点智能评分系统v2.0 @ {report.get('timestamp','')}" ) body = { "name": name, "type": "充电桩热点", "region": region, "address": addr or region, "contactName": "", "contactPhone": "", "contact": "场站热点评分系统", "status": "待开发", "transformerStatus": "待确认", "priority": priority, "note": note, } try: res = crm_post("/api/target-sites", body) rid = (res or {}).get("record", {}).get("id") or (res or {}).get("id") if not rid: return {"ok": False, "name": name, "error": "no id: " + str(res)[:160]} time.sleep(0.3) got = crm_get("/api/target-sites/" + str(rid)) if got and (got.get("id") == rid or got.get("name") == name): return {"ok": True, "id": rid, "name": name, "priority": priority} return {"ok": False, "name": name, "id": rid, "error": "GET verify failed (slice risk)"} except Exception as e: return {"ok": False, "name": name, "error": str(e)[:200]} def push_to_crm_with_dedup(report, sstore): """Upload a scored site, skip if already uploaded (dedup store).""" g = report.get("geocode") or {} inp = report.get("input", {}) nm = inp.get("name") or g.get("formatted") or inp.get("address") or "未命名场站" district = g.get("district", "") region = (g.get("city", "") + ("·" + district if district else "")) or "未定位" key = site_upload_key(nm, region) if key in sstore: print(f" [skip] 已上传过 (CRM id {sstore[key].get('id')})") return {"ok": True, "skipped": True, "name": nm} res = upload_report_to_crm(report) if res.get("ok"): sstore[key] = {"id": res.get("id"), "name": nm, "priority": res.get("priority"), "uploaded_at": now_str()} save_site_store(sstore) print(f" OK 已上传 CRM: {nm} -> id {res.get('id')} (优先级 {res.get('priority')})") else: print(f" !! 上传失败: {nm} -> {res.get('error')}") return res def main(): args = sys.argv[1:] dry_run = "--dry-run" in args upload = "--upload" in args skip_tencent = "--skip-tencent" in args limit = None for j in range(len(args) - 1): if args[j] == "--limit": try: limit = int(args[j + 1]) except: limit = None address = lat = lng = name = None cost_ov = {} i = 0 while i < len(args): a = args[i] if a == "--lat" and i+1 < len(args): lat = args[i+1]; i += 2 elif a == "--lng" and i+1 < len(args): lng = args[i+1]; i += 2 elif a == "--name" and i+1 < len(args): name = args[i+1]; i += 2 elif a == "--batch" and i+1 < len(args): run_batch(args[i+1], dry_run, upload, limit, skip_tencent); return elif a.startswith("--"): i += (2 if i+1 < len(args) and not args[i+1].startswith("--") else 1) else: address = a; i += 1 if not address and not (lat and lng): address = "Pingtan Xianranju Inn, Tancheng Town, Donghu Villa, Pingtan County" name = "Pingtan Xianranju Pilot" print(f"\n{'='*60}") print(f"Site Hotspot Scoring System v2.0 | {now_str()}") print(f"{'='*60}") print(f"Input: addr={address} coords=({lat},{lng}) name={name} dry_run={dry_run}") report = evaluate_site(address=address, lat=lat, lng=lng, name=name, cost_overrides=cost_ov if cost_ov else None, dry_run=dry_run, skip_tencent=skip_tencent) if "error" in report: print(f"\nError: {report['error']}") return with open(STORE_FILE, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2, default=str) print(f"\nOK JSON saved: {STORE_FILE}") html = generate_report_html(report) html_path = "site_hotspot_score_report.html" with open(html_path, "w", encoding="utf-8", newline="\n") as f: f.write(html) print(f"OK HTML saved: {html_path}") if upload and not dry_run and report.get("scorer"): print("\n[Upload] 自动上传到场站CRM (crm.ixgn.cn) ...") push_to_crm_with_dedup(report, load_site_store()) sc = report.get("scorer", {}); roi = report.get("roi", {}) print(f"\n{'-'*40}") print(f"Result: {sc.get('total','-')} -> {sc.get('decision','-')}") print(f"ROI: Annual net {roi.get('net_annual_w','')}wan | Payback {roi.get('payback_years','')}y | NPV5Y={roi.get('npv_5y_w','')}wan") def run_batch(csv_path, dry_run=False, upload=False, limit=None, skip_tencent=False): print(f"\nBatch mode: {csv_path} | upload={upload} | limit={limit} | skip_tencent={skip_tencent}") reports = [] sstore = load_site_store() if upload else {} done = 0 try: with open(csv_path, encoding="utf-8-sig") as f: reader = csv.DictReader(f) for row in reader: if limit and done >= limit: print(f" [limit reached] {done} processed, stopping.") break nm = row.get("name", row.get("Name", row.get("SiteName", ""))) region = row.get("region", row.get("Region", "")) ad = row.get("address", row.get("Address", row.get("DetailAddress", ""))) # Build a full geocodable address: region + address (better AMap accuracy) full_ad = (str(region) + " " + str(ad)).strip() if region else (ad or "") lt = row.get("lat", row.get("Lat", "")) ln = row.get("lng", row.get("Lng", "")) print(f"\n--- [{done+1}] Scoring: {nm or full_ad} ---") r = evaluate_site( address=full_ad if full_ad else None, lat=float(lt) if lt else None, lng=float(ln) if ln else None, name=nm, dry_run=dry_run, skip_tencent=skip_tencent) reports.append({"name": nm or full_ad, "score": (r.get("scorer") or {}).get("total"), "decision": (r.get("scorer") or {}).get("decision"), "report": r}) if upload and r.get("scorer"): push_to_crm_with_dedup(r, sstore) done += 1 time.sleep(0.4) except Exception as e: print(f"Batch read error: {e}") return reports print(f"\n{'='*50}") print(f"Batch summary: {len(reports)} sites") for rp in reports: mk = "OK" if (rp["score"] or 0) >= SCORE_GO else ("WARN" if (rp["score"] or 0) >= SCORE_DISCUSS_LO else "NO") print(f" {mk} {rp['name']}: {rp['score']} -> {rp['decision']}") out_path = "site_batch_scores.csv" with open(out_path, "w", encoding="utf-8-sig", newline="") as f: w = csv.writer(f) w.writerow(["Name", "TotalScore", "Decision", "D1_Stations", "D2_Rate", "D3_Fee", "D4_Geo", "NetAnnual(wan)", "Payback(y)", "NPV5y(wan)"]) for rp in reports: sc = (rp.get("report") or {}).get("scorer") or {} ri = (rp.get("report") or {}).get("roi") or {} ds = sc.get("dim_scores", {}) w.writerow([rp["name"], rp["score"], rp["decision"], ds.get("D1_StationCount", ""), ds.get("D2_ChargeRate", ""), ds.get("D3_ServiceFee", ""), ds.get("D4_Geography", ""), ri.get("net_annual_w", ""), ri.get("payback_years", ""), ri.get("npv_5y_w", "")]) print(f"OK Batch saved: {out_path}") return reports if __name__ == "__main__": main()