import duckdb
from pathlib import Path

class FinancialPipeline:
 """Reusable financial data pipeline engine"""
 
 def __init__(self, db_path: str):
 self.con = duckdb.connect(db_path)
 self._create_schema()
 
 def _create_schema(self):
 """Standard schema: shared across all projects"""
 self.con.execute("""
 -- Bank transactions: exported from online banking
 CREATE TABLE IF NOT EXISTS bank_transactions (
 date DATE,
 type VARCHAR,
 category VARCHAR,
 amount DECIMAL(12,2),
 counterparty VARCHAR,
 source VARCHAR
 )
 """)
 
 self.con.execute("""
 -- Invoice data: from invoice management system
 CREATE TABLE IF NOT EXISTS invoices (
 invoice_date DATE,
 invoice_number VARCHAR,
 seller VARCHAR,
 buyer VARCHAR,
 amount DECIMAL(12,2),
 tax_rate DECIMAL(5,4),
 status VARCHAR
 )
 """)
 
 self.con.execute("""
 -- Platform expenses: Alipay / WeChat / bank transfer
 CREATE TABLE IF NOT EXISTS platform_expenses (
 expense_date DATE,
 platform VARCHAR,
 category VARCHAR,
 amount DECIMAL(12,2),
 vendor VARCHAR,
 remark VARCHAR
 )
 """)

-- Unified income view: bank income + collected invoices
CREATE OR REPLACE VIEW v_all_income AS
SELECT 
 date AS trans_date,
 'bank' AS source_type,
 category,
 amount,
 counterparty AS partner
FROM bank_transactions 
WHERE type = 'income'

UNION ALL

SELECT 
 invoice_date AS trans_date,
 'invoice_system' AS source_type,
 'sales' AS category,
 amount,
 buyer AS partner
FROM invoices 
WHERE status = 'collected';

-- Unified expense view: bank expenses + platform expenses
CREATE OR REPLACE VIEW v_all_expense AS
SELECT 
 date AS trans_date,
 'bank' AS source_type,
 category,
 amount,
 counterparty AS partner
FROM bank_transactions 
WHERE type = 'expense'

UNION ALL

SELECT 
 expense_date AS trans_date,
 'platform' AS source_type,
 category,
 amount,
 vendor AS partner
FROM platform_expenses;

-- Monthly P&L statement
CREATE OR REPLACE VIEW v_monthly_pl AS
WITH monthly_income AS (
 SELECT 
 DATE_TRUNC('month', trans_date) AS month,
 SUM(amount) AS total_income,
 COUNT(*) AS transaction_count
 FROM v_all_income
 GROUP BY 1
),
monthly_expense AS (
 SELECT 
 DATE_TRUNC('month', trans_date) AS month,
 SUM(amount) AS total_expense,
 COUNT(*) AS transaction_count
 FROM v_all_expense
 GROUP BY 1
),
monthly_tax AS (
 -- Extract estimated taxes from invoices
 SELECT 
 DATE_TRUNC('month', invoice_date) AS month,
 SUM(amount * tax_rate) AS estimated_tax
 FROM invoices
 WHERE status IN ('issued', 'collected')
 GROUP BY 1
)
SELECT 
 i.month,
 ROUND(i.total_income, 2) AS revenue,
 ROUND(e.total_expense, 2) AS expenses,
 ROUND(i.total_income - e.total_expense, 2) AS gross_profit,
 ROUND((i.total_income - e.total_expense) / NULLIF(i.total_income, 0) * 100, 2) AS profit_margin_pct,
 ROUND(COALESCE(t.estimated_tax, 0), 2) AS estimated_tax,
 ROUND(i.total_income - e.total_expense - COALESCE(t.estimated_tax, 0), 2) AS net_profit,
 i.transaction_count + e.transaction_count AS total_transactions
FROM monthly_income i
LEFT JOIN monthly_expense e ON i.month = e.month
LEFT JOIN monthly_tax t ON i.month = t.month;

-- Full P&L with MoM and YoY
SELECT 
 month,
 revenue,
 expenses,
 gross_profit,
 profit_margin_pct,
 -- Previous month (MoM)
 LAG(revenue) OVER w AS prev_month_revenue,
 ROUND(
 (revenue - LAG(revenue) OVER w) 
 / NULLIF(LAG(revenue) OVER w, 0) * 100, 2
 ) AS mom_growth_pct,
 -- Same month last year (YoY)
 LAG(revenue, 12) OVER w AS same_month_last_year,
 ROUND(
 (revenue - LAG(revenue, 12) OVER w) 
 / NULLIF(LAG(revenue, 12) OVER w, 0) * 100, 2
 ) AS yoy_growth_pct
FROM v_monthly_pl
WINDOW w AS (ORDER BY month);

class FinancialPipeline:
 """Financial data pipeline engine — write once, use N times"""
 
 def __init__(self, db_path: str):
 self.con = duckdb.connect(db_path)
 self._setup_views()
 
 def _setup_views(self):
 """View consolidation layer: shared across all projects"""
 # v_all_income and v_all_expense from above
 pass
 
 def ingest(self, month: str, data_dir: Path):
 """Data ingestion: called once per month"""
 for csv_file in data_dir.glob(f"{month}*.csv"):
 table_name = csv_file.stem
 self.con.execute(f"""
 COPY {table_name} FROM '{csv_file}' 
 (FORMAT CSV, HEADER, AUTO_DETERMINE)
 """)
 print(f"✅ {month} data ingested")
 
 def generate_report(self, month: str) -> pd.DataFrame:
 """Generate monthly P&L statement"""
 return self.con.execute(f"""
 SELECT * FROM v_monthly_pl 
 WHERE month = '{month}-01'::DATE
 """).fetchdf()
 
 def export_parquet(self, month: str, output_dir: Path):
 """Export as Parquet — 5x faster subsequent queries"""
 df = self.generate_report(month)
 df.to_parquet(output_dir / f"pl_{month}.parquet")
 
 def close(self):
 self.con.close()

# Client A: Restaurant
pipeline_a = FinancialPipeline("clients/restaurant_01.duckdb")
pipeline_a.ingest("2026-07", Path("clients/restaurant_01/data"))
report_a = pipeline_a.generate_report("2026-07")
pipeline_a.export_parquet("2026-07", Path("clients/restaurant_01/output"))
pipeline_a.close()

# Client B: E-commerce — same engine, different data
pipeline_b = FinancialPipeline("clients/ecommerce_02.duckdb")
pipeline_b.ingest("2026-07", Path("clients/ecommerce_02/data"))
report_b = pipeline_b.generate_report("2026-07")
pipeline_b.close()

-- Data quality check view
CREATE OR REPLACE VIEW v_data_quality AS
SELECT 
 'bank_transactions' AS table_name,
 COUNT(*) AS total_rows,
 SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) AS null_amounts,
 SUM(CASE WHEN amount < 0 THEN 1 ELSE 0 END) AS negative_amounts,
 SUM(amount) AS total_amount
FROM bank_transactions

UNION ALL

SELECT 
 'invoices' AS table_name,
 COUNT(*) AS total_rows,
 SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END),
 SUM(CASE WHEN amount < 0 THEN 1 ELSE 0 END),
 SUM(amount)
FROM invoices;