"""
This is by no means a complete implementation of on-disk caching.
It however serves its purpose for my specific need.
"""

lazy import re

lazy import pandas as pd
lazy from bs4 import BeautifulSoup
lazy from edgar import Company

lazy from .cache import persistent

CONCEPTS = {
    "revenue": [
        "RevenueFromContractWithCustomerExcludingAssessedTax",
        "SalesRevenueNet",
        "Revenues",
    ],
    "net_income": [
        "NetIncomeLoss",
        "ProfitLoss",
        "NetIncomeLossAvailableToCommonStockholdersBasic",
    ],
    "eps_basic": ["EarningsPerShareBasic"],
    "eps_diluted": ["EarningsPerShareDiluted"],
    "total_assets": ["Assets"],
}

MONEY = r"\$\s?(\d[\d,]*\.?\d*)\s*(million|billion|M|B)\b"
ARR_RE = [
    re.compile(rf"Cloud\s+ARR[^$.]{{0,150}}?{MONEY}", re.I),
    re.compile(rf"{MONEY}\s+(?:of|in)\s+Cloud\s+ARR", re.I),
]


def clean_facts(sub: pd.DataFrame) -> pd.DataFrame:
    df = sub.dropna(subset=["period_end", "numeric_value"]).copy()
    for dim_col in ["dimensions", "segments", "dimension"]:
        if dim_col in df.columns:
            df = df[df[dim_col].isna() | (df[dim_col] == "") | (df[dim_col] == "{}")]
    return df.sort_values("filing_date")


def parse_flow_concept(sub: pd.DataFrame) -> pd.Series:
    sub = clean_facts(sub)
    if sub.empty:
        return pd.Series(dtype=float)

    sub = sub.dropna(subset=["period_start"]).copy()
    sub["days"] = (sub["period_end"] - sub["period_start"]).dt.days

    sub = sub[sub["days"].between(70, 380)]

    sub = sub.sort_values("filing_date").drop_duplicates(
        ["period_start", "period_end"], keep="last"
    )

    q_discrete = (
        sub[sub["days"].between(75, 105)]
        .set_index("period_end")["numeric_value"]
        .to_dict()
    )

    ytd_6m = sub[sub["days"].between(165, 200)]
    for _, r in ytd_6m.iterrows():
        pend = r["period_end"]
        if pend not in q_discrete:
            q1_match = [
                v
                for k, v in q_discrete.items()
                if 70 <= (pend - k).days <= 105 and k > r["period_start"]
            ]
            if q1_match:
                q_discrete[pend] = r["numeric_value"] - q1_match[0]

    ytd_9m = sub[sub["days"].between(255, 290)]
    for _, r in ytd_9m.iterrows():
        pend = r["period_end"]
        if pend not in q_discrete:
            in_range_q = [
                v for k, v in q_discrete.items() if r["period_start"] < k < pend
            ]
            if len(in_range_q) >= 2:
                q_discrete[pend] = r["numeric_value"] - sum(in_range_q[-2:])

    fy_12m = sub[sub["days"].between(350, 380)]
    for _, r in fy_12m.iterrows():
        pend = r["period_end"]
        if pend not in q_discrete:
            inner_q = [v for k, v in q_discrete.items() if r["period_start"] < k < pend]
            if len(inner_q) == 3:
                q_discrete[pend] = r["numeric_value"] - sum(inner_q)

    res = pd.Series(q_discrete).sort_index()
    res.index = pd.to_datetime(res.index)
    return res


def parse_instant_concept(sub: pd.DataFrame) -> pd.Series:
    sub = clean_facts(sub)
    if sub.empty:
        return pd.Series(dtype=float)

    sub = sub.sort_values(["period_end", "numeric_value"])
    series = sub.drop_duplicates("period_end", keep="last").set_index("period_end")[
        "numeric_value"
    ]
    series.index = pd.to_datetime(series.index)

    full_idx = pd.date_range(
        start=series.index.min(), end=series.index.max(), freq="QE"
    )
    series = series.reindex(series.index.union(full_idx)).sort_index()
    return series.ffill().loc[full_idx]


def extract_revenue(facts: pd.DataFrame) -> pd.Series:
    rev_series = pd.Series(dtype=float)
    for tag in [
        "RevenueFromContractWithCustomerExcludingAssessedTax",
        "SalesRevenueNet",
        "Revenues",
    ]:
        sub = facts[facts["concept"].astype(str).str.endswith(tag)]
        if not sub.empty:
            parsed = parse_flow_concept(sub)
            valid = parsed[parsed > 300e6]
            rev_series = rev_series.combine_first(valid)

    prod_sub = facts[facts["concept"].astype(str).str.endswith("SalesRevenueGoodsNet")]
    serv_sub = facts[
        facts["concept"].astype(str).str.endswith("SalesRevenueServicesNet")
    ]

    if not prod_sub.empty and not serv_sub.empty:
        p_rev = parse_flow_concept(prod_sub)
        s_rev = parse_flow_concept(serv_sub)
        combined_ps = (p_rev + s_rev).dropna()
        rev_series = rev_series.combine_first(combined_ps)

    return rev_series.sort_index()


def cloud_arr(ticker: str, since: str) -> pd.DataFrame:
    rows = {}
    for f in sorted(
        Company(ticker).get_filings(form="8-K", filing_date=f"{since}:"),
        key=lambda f: f.filing_date,
    ):
        if "2.02" not in str(f.items or ""):
            continue
        text = " ".join(
            re.sub(
                r"\s+", " ", BeautifulSoup(a.download(), "html.parser").get_text(" ")
            )
            for a in f.attachments
            if str(a.document_type).upper().startswith("EX-99")
        )
        for pat in ARR_RE:
            if m := pat.search(text):
                num, unit = m.groups()[-2:]
                multiplier = 1e9 if unit[0].lower() == "b" else 1e6
                val = float(num.replace(",", "")) * multiplier
                rows[pd.Timestamp(f.filing_date) - pd.offsets.QuarterEnd(1)] = {
                    "cloud_arr": val,
                    "filing_date": f.filing_date,
                    "snippet": text[max(0, m.start() - 40) : m.end() + 60],
                }
                break
    return pd.DataFrame.from_dict(rows, orient="index").sort_index()


@persistent
def get_financials(ticker: str = "TDC", start: str = "2011-01-01") -> pd.DataFrame:
    facts = Company(ticker).get_facts().to_dataframe(pit_mode=True)
    for c in ("period_start", "period_end", "filing_date"):
        if c in facts.columns:
            facts[c] = pd.to_datetime(facts[c], errors="coerce")

    series = {}

    asset_facts = facts[facts["concept"].astype(str).str.endswith("Assets")]
    series["total_assets"] = parse_instant_concept(asset_facts)

    series["revenue"] = extract_revenue(facts)

    for name in ["net_income", "eps_basic", "eps_diluted"]:
        concept_candidates = CONCEPTS[name]
        combined = pd.Series(dtype=float)
        for tag in concept_candidates:
            sub = facts[facts["concept"].astype(str).str.endswith(tag)]
            if not sub.empty:
                combined = combined.combine_first(parse_flow_concept(sub))
        series[name] = combined

    out = pd.DataFrame(series).sort_index()

    out = out.resample("QE").last()

    out["total_assets"] = out["total_assets"].ffill()
    out["avg_assets"] = out["total_assets"].rolling(2).mean()
    out["roa_quarterly"] = out["net_income"] / out["avg_assets"]
    out["roa_annualized"] = out["roa_quarterly"] * 4

    arr = cloud_arr(ticker, since="2020-01-01")
    if not arr.empty:
        arr.index = pd.to_datetime(arr.index).to_period("Q").to_timestamp("Q")
        out["cloud_arr"] = arr["cloud_arr"]
        out["cloud_arr_yoy"] = out["cloud_arr"].pct_change(4)

    df = out[out.index >= start].copy()
    df.index.name = "quarter_end"
    return df


@persistent
def get_periodic_filing_dates(
    ticker: str = "TDC", start: str = "2011-01-01"
) -> pd.DataFrame:
    company = Company(ticker)
    filings = company.get_filings(
        form=["10-K", "10-Q"],
        filing_date=f"{start}:",
        amendments=False,
    )

    df = filings.to_pandas()
    df.columns = [c.lower() for c in df.columns]

    df["filing_date"] = pd.to_datetime(df["filing_date"])
    df = df.sort_values("filing_date").reset_index(drop=True)
    return df[["filing_date", "form"]]