from pyspark.sql import functions as F
from pyecharts.charts import Pie
from pyecharts import options as opts
from pyecharts.commons.utils import JsCode
import pickle
from 大数据.读取Hbase函数 import HBaseDataLoading

def create_pie():
    # 需要的列族:列限定符
    column = ['info:地址', 'info:主要外资来源国别或地区代码', 'info:主要外资出资比例']

    df = HBaseDataLoading(column)

    # 过滤出"主要外资来源国别或地区代码"列不为0的行
    df_filtered = df.filter(F.col("主要外资来源国别或地区代码") != 0)

    # 将Spark DataFrame转换为Pandas DataFrame
    pandas_df = df_filtered.toPandas()

    # 计算每个主要外资来源国别或地区代码的频次
    country_counts = pandas_df['主要外资来源国别或地区代码'].value_counts()

    print(country_counts)
    # 可视化

    # 创建Pie对象
    pie = Pie()

    # 创建字典，键是主要外资来源国别或地区代码，值是一个包含所有地址:主要外资出资比例的列表
    tooltip_dict = pandas_df.groupby('主要外资来源国别或地区代码').apply(
        lambda x: x[['地址', '主要外资出资比例']].values.tolist()).to_dict()

    # 创建一个JavaScript函数，这个函数会根据当前的主要外资来源国别或地区代码来生成一个包含所有地址:主要外资出资比例的字符串
    js_func = """
    function (params) {{
        var data = {0};
        var country_code = params.name;
        var tooltip_str = '';
        for (var i = 0; i < data[country_code].length; i++) {{
            tooltip_str += '地址: ' + data[country_code][i][0] + ', 主要外资出资比例: ' + data[country_code][i][1] + '<br>';
        }}
        return tooltip_str;
    }}
    """.format(tooltip_dict)

    # 添加数据，主要外资来源国别或地区代码作为x轴，频次作为y轴
    pie.add(
        "",
        [list(z) for z in zip(country_counts.index.tolist(), country_counts.tolist())],
        radius=["10%", "20%"],  # 控制内外半径，即图的大小
        center=["75%", "220px"],  # 饼图中心的位置，即整个图的位置。格式：[左右，上下]
    )

    # 设置系列配置项
    pie.set_series_opts(tooltip_opts=opts.TooltipOpts(formatter=JsCode(js_func)),
                        label_opts=opts.LabelOpts(formatter="{b}: {d}%"))

    return pie