#!/usr/bin/env python3 """AllFunds Dashboard — HTTP server with no-cache headers + /notes API + /api/top10""" import http.server, os, json, shutil, re, subprocess from urllib.parse import urlparse, parse_qs from datetime import datetime # ── Top-10 helper ── GOG_ENV = {'GOG_KEYRING_PASSWORD': 'sigi_gog_2026', 'GOG_ACCOUNT': 'derekchen@siegfriedfund.com', **os.environ} TOP10_CACHE_DIR = '/tmp/top10_cache' def _parse_top10_xlsx(xlsx_path, data_month): import openpyxl wb = openpyxl.load_workbook(xlsx_path, data_only=True) def parse_area(ws): funds = {}; cur = None; rows = [] for r in ws.iter_rows(values_only=True): b,c,d,e = (r[i] if len(r)>i else None for i in (1,2,3,4)) skip = {'Asia','Europe','U.S.A.','UK','Canada','TOTAL',None,''} if b and str(b).strip() not in skip and c is None: if cur and rows: funds[cur]=rows cur=str(b).strip(); rows=[] elif cur and b and str(b).strip()=='TOTAL' and c is None: rows.append({'area':'TOTAL','sub':'','amt':float(d) if d else 0,'pct':float(e) if e else 0}) elif cur and b: rows.append({'area':str(b).strip(),'sub':str(c).strip() if c else '','amt':float(d) if d else 0,'pct':float(e) if e else 0}) if cur and rows: funds[cur]=rows return funds def parse_type(ws): cats={'Bank','Corporate','Insurance Companies','Pension Funds','HNWI','Fund Managers','Endowment','Trust'} funds={}; cur=None; rows=[] for r in ws.iter_rows(values_only=True): b,c,d = (r[i] if len(r)>i else None for i in (1,2,3)) if b and str(b).strip() not in cats and str(b).strip()!='TOTAL' and c is None: if cur and rows: funds[cur]=rows cur=str(b).strip(); rows=[] elif cur and b and str(b).strip()=='TOTAL': rows.append({'type':'TOTAL','amt':float(c) if c else 0,'pct':float(d) if d else 0}) elif cur and b and str(b).strip() in cats: rows.append({'type':str(b).strip(),'amt':float(c) if c else 0,'pct':float(d) if d else 0}) if cur and rows: funds[cur]=rows return funds area_data = parse_area(wb['Area']) type_data = parse_type(wb['Type']) area_grand = next((r[3] for r in wb['Area'].iter_rows(values_only=True) if r[1] and str(r[1]).strip()=='TOTAL' and len(r)>3 and r[3]),0) type_grand = next((r[2] for r in wb['Type'].iter_rows(values_only=True) if r[1] and str(r[1]).strip()=='TOTAL' and len(r)>2 and r[2]),0) wb.close() return {'data_month':data_month,'area':area_data,'area_grand_total':float(area_grand),'type':type_data,'type_grand_total':float(type_grand)} def _fetch_top10(data_month): """Download + parse top10 for given YYYY-MM. Returns dict or raises.""" os.makedirs(TOP10_CACHE_DIR, exist_ok=True) cache_json = f'{TOP10_CACHE_DIR}/top10_{data_month}.json' if os.path.exists(cache_json): with open(cache_json,'r',encoding='utf-8') as f: return json.load(f) # Compute search prefix: data_month + 1 month dt = datetime.strptime(data_month, '%Y-%m') next_m = dt.month % 12 + 1 next_y = dt.year + (1 if dt.month == 12 else 0) prefix = f'{str(next_y)[2:]}{next_m:02d}' # e.g. '2609' # Search Drive # 搜「top 10 investors」，再過濾 prefix（YYMM，如 2610） r = subprocess.run(['/usr/local/bin/gog','--client','derek','drive','search', 'top 10 investors','--plain'], capture_output=True, text=True, timeout=60, env=GOG_ENV) lines = [l for l in r.stdout.splitlines() if 'top 10 investors' in l.lower() and prefix in l and '.xlsx' in l.lower()] if not lines: raise FileNotFoundError(f'No file found for {data_month} (prefix {prefix})') lines.sort(reverse=True) # latest file first file_id = lines[0].split()[0].strip() file_name = re.sub(r'^\S+\s+', '', lines[0]).rsplit(None, 3)[0].strip() xlsx_path = f'{TOP10_CACHE_DIR}/top10_{data_month}.xlsx' subprocess.run(['/usr/local/bin/gog','--client','derek','drive','download', file_id,'--output',xlsx_path], capture_output=True, text=True, timeout=120, env=GOG_ENV) data = _parse_top10_xlsx(xlsx_path, data_month) data['source_file'] = file_name with open(cache_json,'w',encoding='utf-8') as f: json.dump(data,f,ensure_ascii=False) return data # ── End Top-10 helper ── NOTES_FILE = '/tmp/notes_data.json' class NoCacheHandler(http.server.SimpleHTTPRequestHandler): def do_OPTIONS(self): self.send_response(200) self.end_headers() def do_GET(self): parsed = urlparse(self.path) if parsed.path == '/notes': self._serve_notes() return if parsed.path == '/api/top10': qs = parse_qs(parsed.query) month = qs.get('month', [None])[0] self._serve_top10(month) return super().do_GET() def do_POST(self): if self.path == '/notes': self._save_notes() return self.send_response(404) self.end_headers() def _json_response(self, body, status=200): enc = body.encode('utf-8') if isinstance(body, str) else body self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(enc))) self.end_headers() self.wfile.write(enc) def _serve_top10(self, month): if not month or not re.match(r'^\d{4}-\d{2}$', month): self._json_response('{"error":"Invalid month. Use YYYY-MM format."}', 400) return try: data = _fetch_top10(month) self._json_response(json.dumps(data, ensure_ascii=False)) except FileNotFoundError as e: self._json_response(json.dumps({'error': str(e)}), 404) except Exception as e: self._json_response(json.dumps({'error': str(e)}), 500) def _serve_notes(self): try: if os.path.exists(NOTES_FILE): with open(NOTES_FILE, 'r', encoding='utf-8') as f: data = f.read() else: data = '{}' self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(data.encode('utf-8')) except Exception: self.send_response(500) self.end_headers() def _save_notes(self): try: length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length) parsed = json.loads(body) with open(NOTES_FILE, 'w', encoding='utf-8') as f: json.dump(parsed, f, ensure_ascii=False, indent=None) self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"ok":true}') except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(('{"ok":false,"error":"'+str(e)+'"}').encode()) def end_headers(self): self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate') self.send_header('Pragma', 'no-cache') self.send_header('Expires', '0') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') super().end_headers() def log_message(self, fmt, *args): pass # suppress access logs os.chdir('/tmp') # Ensure bb_alert_links.json is available (always refresh from workspace for HTTPS URLs) _src = '/root/.openclaw-cooper/workspace/dashboard/data/bb_alert_links.json' _dst = '/tmp/bb_alert_links.json' if os.path.exists(_src): shutil.copy(_src, _dst) # Always overwrite to ensure HTTPS URLs are used elif not os.path.exists(_dst): pass # No source, no destination — skip print('HTTP server: http://143.198.88.210:18999/ (+ /notes API)') http.server.HTTPServer(('', 18999), NoCacheHandler).serve_forever()