import os
from datetime import datetime, timedelta
import sys
from typing import List
import lark_oapi as lark
from environs import Env
from langlink_client import execute_langlink_app
from category_block import create_chart_block, create_raw_figure_block
from nest_block import NestBlock
from md_block import create_markdown_block, create_text_block
from report import Report
import pandas as pd
from matplotlib.ticker import FuncFormatter

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from render import Render

env = Env()
env.read_env()

LANGLINK_APP_ID = env("LANGLINK_REPORT_APP_ID")

def run_billing_summary(content: str):
    return execute_langlink_app(LANGLINK_APP_ID, content)

r = Render(os.path.join(os.path.dirname(__file__), '..', 'weekly.yaml'), env.str("TRINO_HOST"))

today = datetime.now().strftime('%B %d, %Y')
begin_date = (datetime.now() - timedelta(days=10)).strftime('%B %d, %Y')
end_date = (datetime.now() - timedelta(days=4)).strftime('%B %d, %Y')
chart_date = (datetime.now() - timedelta(days=17)).strftime('%B %d, %Y')

def render_description(r: Render):
    return create_markdown_block(f'''- Analysis of Time Range
  - Current billing period: {begin_date} - {end_date}
  - Analysis date: {today}
  - Time range of screenshots: {chart_date} - {end_date}
- Filtering Criteria
  - Exclude relevant Serverless bills (meaning the data does not include Serverless costs).
  - Exclude internal tenants (meaning the data does not include costs related to PingCAP-Prod (Ossingisht related tenant)).
  - Retain only On_Demand status tenants (meaning the data does not include costs for tenants in states like POC).
  - Exclude credit discount (meaning the data does not include credits provided for business reasons).
- Billing Values
  - The billing data analyzed in this report includes values after applying discounts but without deducting credits. This means the value is the Running Total - Discount, as seen on the Bills page. (If the tenant does not have credits, this value equals Total Cost).
    - Because discounts are averaged over each day, this deduction does not affect the overall user\'s daily cost curve.
    - Once credits are provided to the user, they are deducted within 4-5 days, impacting the user\'s daily cost curve and not accurately reflecting their usage.
- Data Delay
  - The data source (bigquery) has a 3-day delay.
- Basic Concepts
  - WoW (Week-over-Week): The change in data from the current period [t-7, t] compared to the previous period [t-14, t-6].
- The data is provided by the Digital Foundation Team.
    ''')

def render_total_daily(r: Render):
    return create_markdown_block(run_billing_summary(f'''
The overall bill trend for the last 7 days ({begin_date} - {end_date}):
- Average bill for the past 7 days: ${r.sum_avg:,.2f}, compared to the previous period (${r.prev_sum_avg:,.2f}): {r.sum_avc_inc_percent:+.2f}%
- Maximum value: On {r.max_daily_cost_date}, ${r.max_daily_cost_value:,.2f}, compared to the previous period (${r.prev_max_daily_cost_value:,.2f}): {(r.max_daily_cost_value-r.prev_max_daily_cost_value)/r.prev_max_daily_cost_value*100:+.2f}%. Minimum value: On {r.min_daily_cost_date}, ${r.min_daily_cost_value:,.2f}, compared to the previous period (${r.prev_min_daily_cost_value:,.2f}): {(r.min_daily_cost_value-r.prev_min_daily_cost_value)/r.prev_min_daily_cost_value*100:+.2f}%.
- Daily bill details for the last seven days:
{r.sum_cost_every_day}
'''))

def render_daily_every_tenant(r: Render):
    daily_every_tenant_trend = f'''The billing trend for subtenants in the last 7 days ({begin_date} - {end_date}):
- A total of {r.paid_tenant_count} tenants have bills, with an increase of {r.paid_tenant_inc_count:+} compared to the previous period.
'''

    if len(r.dataframe('leave_paid_customers')) > 0:
        daily_every_tenant_trend += '  - The tenants that decreased are:' + ', '.join(r.dataframe('leave_paid_customers').tenant_name) + '\n'
    if len(r.dataframe('new_paid_customers')) > 0:
        daily_every_tenant_trend += '  - The tenants that increased are:' + ', '.join(r.dataframe('new_paid_customers').tenant_name) + '\n'

    customer_analyze = ''
    with r.context(tenant_id=r.biggest_customer_id):
        customer_analyze += f'''- The tenant with the highest proportion is: {r.customer_name}
  - Daily average: {r.customer_daily_avg_percent:.2f}%, End-of-period to Beginning-of-period change: {r.customer_daily_percent_diff:+.2f}%
  - Maximum value: On {r.customer_max_daily_percent_date}, {r.customer_max_daily_percent_value:.2f}%, Minimum value: On {r.customer_min_daily_percent_date}, {r.customer_min_daily_percent_value:.2f}%
'''

    if r.biggest_customer_id != r.potential_customer_id:
        with r.context(tenant_id=r.potential_customer_id):
            customer_analyze += f'''- The tenant with the fastest growth is: {r.customer_name}
  - Daily average: {r.customer_daily_avg_percent:.2f}%, End-of-period to Beginning-of-period change: {r.customer_daily_percent_diff:+.2f}%
  - Maximum value: On {r.customer_max_daily_percent_date}, {r.customer_max_daily_percent_value:.2f}%, Minimum value: On {r.customer_min_daily_percent_date}, {r.customer_min_daily_percent_value:.2f}%
'''

    customer_analyze += '''- Billing status for new tenants (created within the last 180 days) in the last 7 days:
'''
    for _, row in r.dataframe('new_custermers').iterrows():
        with r.context(tenant_id=row.tenant_id):
            customer_analyze += f'''  - {r.customer_name}, Creation Time: {row.tenant_created_at}, Average Percentage: {r.customer_sum_percent:.2f}%, Compared to the Previous Period ({r.customer_prev_sum_percent:.2f}%): {r.customer_sum_percent-r.customer_prev_sum_percent:+.2f}%
'''

    daily_every_tenant = [
        render_cost_by_tenant_chart(r),
        create_markdown_block(run_billing_summary(daily_every_tenant_trend + customer_analyze))
    ]

    return daily_every_tenant

def render_top5tenant(r: Render):
    top5tenant = [
        create_markdown_block('Top 5 tenants with the highest bills in the last 7 days:'),
    ]

    for _, row in r.dataframe('top_5_customers').iterrows():
        with r.context(tenant_id=row.billing_tenant_id):
            customer_missing = ''
            if len(r.dataframe('customer_missing_projects')) > 0:
                customer_missing += '  - The projects that decreased are:' + ', '.join(r.dataframe('customer_missing_projects').billing_project_name)

            top5tenant = top5tenant + [
                create_markdown_block(f'''## {r.customer_name}
### Daily bills
'''),
                create_chart_block('customer_sum_cost_every_day', {
                    "type": "line",
                    "x": "billing_cost_date",
                    "y": "daily_cost"
                }, r),
                create_markdown_block(run_billing_summary(f'''The overall billing trend for the last 7 days ({begin_date} - {end_date}):
- Average bill for the past 7 days: ${r.customer_sum_avg:,.2f}, compared to the previous period (${r.prev_customer_sum_avg:,.2f}): {(r.customer_sum_avg-r.prev_customer_sum_avg)/r.prev_customer_sum_avg*100:+.2f}%
- Maximum value: On {r.customer_max_daily_cost_date}, ${r.customer_max_daily_cost_value:,.2f}, compared to the previous period (${r.prev_customer_max_daily_cost_value:,.2f}): {(r.customer_max_daily_cost_value-r.prev_customer_max_daily_cost_value)/r.prev_customer_max_daily_cost_value*100:+.2f}%. Minimum value: On {r.customer_min_daily_cost_date}, ${r.customer_min_daily_cost_value:,.2f}, compared to the previous period (${r.prev_customer_min_daily_cost_value:,.2f}): {(r.customer_min_daily_cost_value-r.prev_customer_min_daily_cost_value)/r.prev_customer_min_daily_cost_value*100:+.2f}%.
''')),
                create_markdown_block('### Daily project bills'),
            ]

            daily_project_bills = f'''
The trend for each project in the last 7 days ({begin_date} - {end_date}):
- A total of {r.customer_billing_project_count} projects incurred expenses, with a change of {r.customer_billing_project_count-r.customer_prev_billing_project_count:+} compared to the previous period ({r.customer_prev_billing_project_count}).
{customer_missing}
- Top 3 Projects
'''
            daily_project_bills += render_top3project(r)

            if len(r.dataframe('customer_new_projects')) > 0:
                daily_project_bills += '- New Projects\n'
                for _, prow in r.dataframe('customer_new_projects').iterrows():
                    daily_project_bills += '\n'
                    with r.context(project_id=prow.billing_project_id):
                        if not r.project_sum_avg:
                            continue
                        daily_project_bills += f'  - Project Name: {prow.billing_project_name}\n'
                        daily_project_bills += f'    - 7-day average bill: ${r.project_sum_avg:,.2f}\n'
                        daily_project_bills += f'    - Maximum value: On {r.project_max_daily_cost_date}, ${r.project_max_daily_cost_value:,.2f}, '
                        daily_project_bills += f'Minimum value: On {r.project_min_daily_cost_date}, ${r.project_min_daily_cost_value:,.2f}\n'

            if len(r.dataframe('customer_disabled_projects')) > 0:
                daily_project_bills += '- Disable Projects\n'
                for _, prow in r.dataframe('customer_disabled_projects').iterrows():
                    daily_project_bills += '\n'
                    with r.context(project_id=prow.billing_project_id):
                        daily_project_bills += f'  - Project Name: {prow.billing_project_name}, Cost disappearance date: On {prow.disabled_date}\n'

            top5tenant = top5tenant + [
                create_markdown_block(run_billing_summary(daily_project_bills)),
                create_markdown_block('### Daily service bills'),
            ]

            top_services = '- Top 3 Services\n'
            for idx, srow in r.dataframe('customer_top_3_service').iterrows():
                with r.context(unit=srow.billing_unit):
                    top_services += f'  - No.{idx+1}: {srow.billing_unit}\n'
                    top_services += f'    - 7-day average bill: ${r.customer_unit_sum_avg:,.2f}, compared to the previous period (${r.prev_customer_unit_sum_avg:,.2f})'
                    top_services += f'{(r.customer_unit_sum_avg-r.prev_customer_unit_sum_avg)/r.prev_customer_unit_sum_avg*100:+.2f}%\n'
                    top_services += f'    - Maximum value: On {r.customer_unit_max_daily_cost_date}, ${r.customer_unit_max_daily_cost_value:,.2f},'
                    top_services += f'compared to the previous period (${r.prev_customer_unit_max_daily_cost_value:,.2f})'
                    top_services += f'{(r.customer_unit_max_daily_cost_value-r.prev_customer_unit_max_daily_cost_value)/r.prev_customer_unit_max_daily_cost_value*100:+.2f}%.'
                    top_services += f'Minimum value: On {r.customer_unit_min_daily_cost_date}, ${r.customer_unit_min_daily_cost_value:,.2f},'
                    top_services += f'compared to the previous period (${r.prev_customer_unit_min_daily_cost_value:,.2f})'
                    top_services += f'{(r.customer_unit_min_daily_cost_value-r.prev_customer_unit_min_daily_cost_value)/r.prev_customer_unit_min_daily_cost_value*100:+.2f}%。\n'
                    top_services += f'    - Daily bill details for the last seven days:\n{r.customer_unit_sum_cost_every_day}\n'

            if len(r.dataframe('customer_new_units')) > 0:
                top_services += '- New Services\n'
                for _, srow in r.dataframe('customer_new_units').iterrows():
                    top_services += '\n'
                    with r.context(unit=srow.billing_unit):
                        top_services += f'  - Service Name: {srow.billing_unit}, Occurrence Time: On {srow.billing_cost_date}\n'
                        top_services += f'    - 7-day average bill: ${r.customer_unit_sum_avg:,.2f}.\n'
                        top_services += f'    - Maximum value: On {r.customer_unit_max_daily_cost_date}, ${r.customer_unit_max_daily_cost_value:,.2f}.'
                        top_services += f'Minimum value: On {r.customer_unit_min_daily_cost_date}, ${r.customer_unit_min_daily_cost_value:,.2f}.\n'

            if len(r.dataframe('customer_disabled_units')) > 0:
                top_services += '- Disabled Services\n'
                for _, srow in r.dataframe('customer_disabled_units').iterrows():
                    top_services += '\n'
                    with r.context(unit=srow.billing_unit):
                        top_services += f'  - Service Name: {srow.billing_unit}, Cost disappearance time: On {srow.disabled_date}\n'

            top5tenant = top5tenant + [
                create_markdown_block(run_billing_summary(top_services)),
                create_markdown_block('### Every Project\'s Bills'),
            ]

            projects_bills = ''
            for _, prow in r.dataframe('customer_top_3_projects').iterrows():
                projects_bills += '\n'
                with r.context(project_id=prow.billing_project_id):
                    projects_bills += f'#### {prow.billing_project_name}\n'
                    projects_bills += '- Top 3 Service\n'
                    for idx, srow in r.dataframe('project_top_3_service').iterrows():
                        with r.context(unit=srow.billing_unit):
                            projects_bills += f'  - No.{idx+1}: Service Name: {srow.billing_unit}\n'
                            projects_bills += f'    - 7-day average bill: ${r.project_unit_sum_avg:,.2f}, compared to the previous period (${r.prev_project_unit_sum_avg:,.2f})'
                            projects_bills += f'{(r.project_unit_sum_avg-r.prev_project_unit_sum_avg)/r.prev_project_unit_sum_avg*100:+.2f}%\n'
                            projects_bills += f'    - Maximum value: On {r.project_unit_max_daily_cost_date}, ${r.project_unit_max_daily_cost_value:,.2f},'
                            projects_bills += f'compared to the previous period(${r.prev_project_unit_max_daily_cost_value:,.2f})'
                            projects_bills += f'{(r.project_unit_max_daily_cost_value-r.prev_project_unit_max_daily_cost_value)/r.prev_project_unit_max_daily_cost_value*100:+.2f}%.'
                            projects_bills += f'Minimum value: On {r.project_unit_min_daily_cost_date}, ${r.project_unit_min_daily_cost_value:,.2f}, compared to the previous period'
                            projects_bills += f'(${r.prev_project_unit_min_daily_cost_value:,.2f}) '
                            projects_bills += f'{(r.project_unit_min_daily_cost_value-r.prev_project_unit_min_daily_cost_value)/r.prev_project_unit_min_daily_cost_value*100:+.2f}%.\n'

                    if len(r.dataframe('project_new_units')) > 0:
                        projects_bills += '- New Services\n'
                        for _, srow in r.dataframe('project_new_units').iterrows():
                            with r.context(unit=srow.billing_unit):
                                projects_bills += f'  - Service Name: {srow.billing_unit}, Occurrence Time: On {srow.billing_cost_date}\n'
                                projects_bills += f'    - 7-day average bill: ${r.project_unit_sum_avg:,.2f}.\n'
                                projects_bills += f'    - Maximum value: On {r.project_unit_max_daily_cost_date}, ${r.project_unit_max_daily_cost_value:,.2f}.'
                                projects_bills += f'Minimum value: On {r.project_unit_min_daily_cost_date}, ${r.project_unit_min_daily_cost_value:,.2f}.\n'

                    if len(r.dataframe('project_disabled_units')) > 0:
                        projects_bills += '- Disabled Services\n'
                        for _, srow in r.dataframe('project_disabled_units').iterrows():
                            with r.context(unit=srow.billing_unit):
                                projects_bills += f'  - Service Name: {srow.billing_unit}, Cost disappearance time: On {srow.disabled_date}\n'

            top5tenant = top5tenant + [
                create_markdown_block(run_billing_summary(projects_bills)),
            ]

    return top5tenant

def render_top3project(r: Render):
    top3project = ''
    for idx, prow in r.dataframe('customer_top_3_projects').iterrows():
        with r.context(project_id=prow.billing_project_id):
            top3project += f'''  - No.{idx+1}: Project Name: {prow.billing_project_name}
    - Average bill for the projects in the past 7 days: ${r.project_sum_avg:,.2f}, compared to the previous period (${r.project_prev_sum_avg:,.2f}): {(r.project_sum_avg-r.project_prev_sum_avg)/r.project_prev_sum_avg*100:+.2f}%
    - Maximum value: On {r.project_max_daily_cost_date}, ${r.project_max_daily_cost_value:,.2f}, compared to the previous period (${r.prev_project_max_daily_cost_value:,.2f}): {(r.project_max_daily_cost_value-r.prev_project_max_daily_cost_value)/r.prev_project_max_daily_cost_value*100:+.2f}%. Minimum value: On {r.project_min_daily_cost_date}, ${r.project_min_daily_cost_value:,.2f}, compared to the previous period (${r.prev_project_min_daily_cost_value:,.2f}): {(r.project_min_daily_cost_value-r.prev_project_min_daily_cost_value)/r.prev_project_min_daily_cost_value*100:+.2f}%.
    - Daily bill details for the last seven days:
{r.project_sum_cost_every_day}
'''
    return top3project

def percent_formatter(x, _):
    return f'{x:.0%}'

def render_cost_by_tenant_chart(r: Render):
    with r.context(begin_date=(datetime.now() - timedelta(days=17)).strftime('%Y-%m-%d'), end_date=end_query_date):
        df = r.dataframe('sum_cost_every_day_by_tenant')
        df['billing_cost_date'] = pd.to_datetime(df['billing_cost_date'])
        pivot_df = df.pivot_table(index='billing_cost_date', columns='tenant_name', values='percent', aggfunc='sum').fillna(0)
        # cumulative_df = pivot_df.cumsum(axis=1)
        p = pivot_df.plot(kind='area', stacked=True, alpha=0.8, figsize=(16,9), legend=None)
        # plt.subplots_adjust(right=0.75)

        p.yaxis.set_major_formatter(FuncFormatter(percent_formatter))

        threshold_percent = 3  # 根据需要调整阈值百分比
        annotated_tenants = set()
        for date in pivot_df.index:
            for column in pivot_df.columns:
                if column not in annotated_tenants:
                    percent_at_date = pivot_df.loc[date, column]
                    if percent_at_date > threshold_percent / 100:
                        p.annotate(column, xy=(date, percent_at_date), xytext=(date, percent_at_date + 0.02),
                                    arrowprops=dict(facecolor='black', arrowstyle='->'),
                                    fontsize=8, ha='left', va='bottom')
                        annotated_tenants.add(column)

        return create_raw_figure_block(p)

begin_query_date = (datetime.now() - timedelta(days=10)).strftime('%Y-%m-%d')
end_query_date = (datetime.now() - timedelta(days=4)).strftime('%Y-%m-%d')

def render_total_percent_by_tenant(r: Render):
    total_percent_by_tenant: List[NestBlock] = [render_topn_chart(r)]

    with r.context(begin_date=begin_query_date, end_date=end_query_date):
        df = r.dataframe('sum_topn_cost_percent_by_tenant')
        df["Rank"] = df["percentage"].rank(ascending=False)
        df["% Total cost"] = df['percentage']

    with r.context(begin_date=(datetime.now() - timedelta(days=17)).strftime('%Y-%m-%d'), end_date=(datetime.now() - timedelta(days=11)).strftime('%Y-%m-%d')):
        previous_df = r.dataframe('sum_topn_cost_percent_by_tenant')
        previous_df["Rank_previous"] = previous_df["percentage"].rank(ascending=False)

    merged_df = pd.merge(df, previous_df, on="tenant_name", how="outer", suffixes=('', '_previous'))
    merged_df["% Diff"] = merged_df["percentage"] - merged_df["percentage_previous"]
    merged_df["Rank Diff"] = merged_df["Rank_previous"] - merged_df["Rank"]

    result_df = merged_df[["Rank", "Rank Diff", 'tenant_name', '% Total cost', '% Diff']].head(10).to_markdown()

    # dropped_out_df = previous_df[previous_df["Rank_previous"] <= 10 & df["Rank"] >= 10].head(10)

    total_percent_by_tenant.append(create_text_block(result_df))

    return total_percent_by_tenant

def render_topn_chart(r: Render):
    df = r.dataframe('sum_topn_cost_percent_by_tenant')
    p = df.plot(kind='pie', y='percentage', labels=df['tenant_name'], autopct='%1.1f%%', startangle=90, legend=None)
    return create_raw_figure_block(p)

with r.context(begin_date=begin_query_date, end_date=end_query_date):
    description = render_description(r)
    total_daily = render_total_daily(r)
    daily_every_tenant = render_daily_every_tenant(r)
    total_percent_by_tenant = render_total_percent_by_tenant(r)
    top5tenant = render_top5tenant(r)

    report = Report(f'Dedicated Weekly Billing Analysis({begin_date} - {end_date})', env.str("REPORT_SPACE_ID"), env.str("WEEKLY_FOLDER_ID"))
    report.create(os.path.join(os.path.dirname(__file__), 'weekly.yaml'), r, {
        "description": description,
        "total_daily": total_daily,
        "daily_every_tenant": daily_every_tenant,
        "total_percent_by_tenant": total_percent_by_tenant,
        "top5tenant": top5tenant
    })