import panda as pd import yaml,detatime,hashilb,json from pathlib import Path from templating import get_env,rander from models import CompanyMeta,ReportSections,RenderPayload from render import hetml_to_pdf,html_to_docx DISPLAY_NAME={ "co2_emission":"Co2排出量", "energy_renewable_ratio":"再生可能エネルギー比率", "female_management_ratio":"女性管理職比率" } def sha256(p:Path)->str: h=hashlib.sha256() with p.open("rb") as f: for chunk in iter(lambda: f.read(8192), b""): h.update(chunk) return h.hexdigest() def load_company_meta(path:str)->CompanyMeta: data=yaml.safe_load(Path(path).read_text(encoding="utf-8")) return CompanyMeta(**data) def load_financials(path:str)->pd.DataFrame: return pd.read_csv(path) def compute_kpi(fin_df:pd.DataFrame,fiscal_year:int): latest=fin_df[fin_df["year"]==fiscal_year].sort_values("quarter").tail(1) prev=fin_df[fin_df["year"]==fiscal_year-1].sort_values("quarter").tail(1) revenue=float(latest["revenue"].iloc[0]) prev_revenue=float(prev["revenue"].iloc[0]) if not prev.empty else 0 ebit=float(latest["ebit"].iloc[0]) if not latest.empty else 0 net_income=float(latest["net_income"].iloc[0]) if not latest.empty else 0 equity=float(latest["equity"].iloc[0]) if not latest.empty else 0.0 ebit_margin=ebit /revenue *100 if revenue else 0.0 revenue_yoy=((revenue/float(prev["revenue"].iloc[0]))-1)*100 if not prev.empty and float(prev["revenue"].iloc[0]) else 0.0 return { "revenue": revenue, "ebit": ebit, "ebit_margin": ebit_margin, "net_income": net_income, "equity": equity, "revenue_yoy": revenue_yoy } def esg_table (df:pd.DataFrame,fiscal_year:int): dfy=df[df["year"]==fiscal_year].copy rows=[] for _, r in dfy.iterrows(): display = DISPLAY_NAME.get(r["metric"], r["metric"]) rows.append({ "display": display, "value": r["value"], "unit": r.get("unit", ""), "notes": r.get("notes", ""), }) return rows def build_sections(meta:CompanyMeta,kpi:dict,esg_rows:list,llm=None)->ReportSections: if llm: ceo_message = llm.generate_ceo_message(meta.kpi,esg_rows) risk = llm.generate_risk_opportunities(meta.kpi,esg_rows) else: ceo_message = f"[{meta.fiscal_year}]期は、売上成長と収益性の両立に注力しました。" risk = "主要リスクはマクロ環境と規制動向。機会は生成AI活用と脱炭素需要の拡大です。" return ReportSections(ceo_message=ceo_message,risk_opportunity=risk) def_translate_payload(payload:dict,lang:str,llm)->dict: """payload のうち、テキスト項目をtarget langに翻訳。数値は非対象。"""    if not llm or lang =="ja": return payload texts=[] texts.append(payload["section"]["ceo_message"]) texts.append(payload["section"]["risk_opportunity"]) for row in payload["esg_table"]: texts.append(row["display"]) texts.append(row["notes"] or "") texts.append(payload["meta"]["report_title"]) for topic in payload["meta"].get("material_topics", []): texts.append(topic) translated = llm.translate_text(texts, target_lang=lang) it = iter(translated) payload["section"]["ceo_message"] = next(it) payload["section"]["risk_opportunity"] = next(it) for row in payload["esg_table"]: row["display"] = next(it) row["notes"] = next(it) payload["meta"]["report_title"] = next(it) mt = payload["meta"].get("material_topics", []) for i in range(len(mt)): mt[i] = next(it) return payload def generate_report(company_yaml,financials_csv,esg_csv, template_dir,template_name="report.html.j2", out_html="output/report.html",out_pdf="output/report.pdf", out_docx="output/report.docx",lang="ja",llm=None): Path(Path(out_html).parent).mkdir(parents=True, exist_ok=True) meta= load_company_meta(company_yaml) fin= load_financials(financials_csv) esg= load_esg(esg_csv) kpi = compute_kpi(fin, meta.fiscal_year) esg_rows = esg_table(esg, meta.fiscal_year) sections = build_sections(meta, kpi, esg_rows, llm=llm) env = get_env(template_dir) payload = RenderPayload( meta=meta, kpi=kpi, esg_table=esg_rows,kpi=kpi, sections=sections, generated_at=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),lang=lang ).model__dump() payload= _translate_payload_texts(payload, lang=lang, llm=llm) html = render(env, template_name, payload) Path(out_html).write_text(html, encoding="utf-8") html_to_pdf(html, out_pdf) html_to_docx(html, out_docx) meta_json ={ "inputs":{ "company_yaml_sha":_sha256(Path(company_yaml)), "financials_csv_sha":_sha256(Path(financials_csv)), "esg_csv_sha":_sha256(Path(esg_csv)), "lang":lang }, "outputs":{"html":out_html,"pdf":out_pdf,"docx":out_docx}, "template":{"dir":templates_dir,"name":template_name}, "generated_at":datetime.datetime.now().strftime("timespec=sedconds"), } return out_html,out_pdf,out_docx,meta_json