import logging import sys from jinja2 import Environment, FileSystemLoader from loguru import logger from fp.新闻资讯 import News from fp.日线数据入库 import StockDailyMain from fp.板块数据入库 import SectorOpt from fp.涨跌停数据 import LimitList from fp.自选股数据 import OptionalStock from fp.行情指数数据 import AllIndex from fp.资金出入概览 import MoneyFlow from utils.comm import * class AppMain: def __init__(self, trade_date=datetime.now()): # 配置日志输出到文件和控制台 logger.add("log/FPAppMain.log", rotation="500 MB", level="INFO") logger.add(sys.stderr, level="INFO") self.trade_date = trade_date.strftime('%Y%m%d') if if_run(trade_date): # 创建一个文件夹 self.file_name = f'html/{self.trade_date}' create_file(self.file_name) # 各大指数数据 self.index = AllIndex(trade_date=trade_date) # 板块数据 self.sector_opt = SectorOpt(trade_date=trade_date) # 涨跌停 炸板 self.limit_list = LimitList(trade_date=trade_date) # 资金流入流出 self.money_flow = MoneyFlow(trade_date=trade_date) # 日线数据 self.stock_daily = StockDailyMain() # 新闻咨询 self.news = News(trade_date=trade_date) # 我的自选 self.optional_stock = OptionalStock(trade_date=trade_date) def calendar_page_data(self): logger.info(f"{self.trade_date} start--> 开始 JSON 文件生成！") res_df = self.index.calendar_page_json() # 将 DataFrame 转换为 JSON，并写入文件 res_df.to_json('html/json/events.json', orient='records') logger.info(f"{self.trade_date} --> JSON 文件已更新") def init_data(self): """ 每日数据初始化 :return: """ logger.info(f"{self.trade_date} start--> 开始数据初始化！") # 日线数据入库 self.stock_daily.task_data(self.trade_date) # 行业板块 入库 self.sector_opt.save_hy_sector() # 概念板块 入库 self.sector_opt.save_gn_sector() # 行业板块个股 self.sector_opt.stock_by_hy_sector() # 概念板块个股 self.sector_opt.stock_by_gn_sector() logger.info(f"{self.trade_date} end--> 数据初始化完毕！") def create_chart(self): logger.info(f"{self.trade_date} start--> 开始图表页面生成！") # 板块图表 self.sector_opt.industry_generate_chart() self.sector_opt.concept_generate_chart() # 涨跌停 炸板 图表 self.limit_list.limit_list_chart() # 资金出入图表 self.money_flow.money_flow_chart() logger.info(f"{self.trade_date} end--> 图表页面生成完毕！") def create_page(self): logger.info(f"{self.trade_date} start--> 开始数据静态页面生成！") badge, card_deck = self.index.html_page_data() limit_list_table_content = self.limit_list.html_page_data() # 新闻数据 cctv_content, cls_content = self.news.html_page_data() # 首页数据 页面 self.__write_html(template_name="index_template.html", write_html_name="index.html", badge=badge, news_cctv=cctv_content, news_cls=cls_content) # 指数行情 页面 self.__write_html(template_name="zs_data_template.html", write_html_name="zs_data.html", card_deck=card_deck) # 涨跌停 炸板 页面 self.__write_html(template_name="limit_list_template.html", write_html_name="limit_list.html", limit_table=limit_list_table_content) # 我的自选 页面 (这个时间会很长) # optional_stock_pills_tab_content = self.optional_stock.html_page_data() # self.__write_html(template_name="zx_data_template.html", write_html_name="zx_data.html", # pills_tabContent=optional_stock_pills_tab_content) logger.info(f"{self.trade_date} --> 数据静态页面生成完毕！") def __write_html(self, template_name, write_html_name, *args, **kwargs): print(template_name, "---", write_html_name) print(dict(*args, **kwargs)) # 设置 Jinja2 环境 env = Environment(loader=FileSystemLoader('template/')) template = env.get_template(template_name) # 将 DataFrame 数据传递给模板并渲染 rendered_output = template.render(dict(*args, **kwargs)) # 将渲染后的内容写入 HTML 文件 with open(self.file_name + f'/{write_html_name}', 'w', encoding='utf-8') as f: f.write(rendered_output) print("HTML rendered and saved as 'output.html'") def upload_ftp(self): logger.info(f"{self.trade_date} start--> 开始FTP上传！") # ftp上传 # 上传json文件 upload_files_to_ftp( 'html/json', '/htdocs/stock/json', "events.json") # 上传html文件 upload_files_to_ftp( f'html/{self.trade_date}', f'/htdocs/stock/{self.trade_date}') logger.info(f"{self.trade_date} end--> FTP上传完毕！") if __name__ == '__main__': # date_obj = datetime.strptime('20231110', "%Y%m%d") app = AppMain() # app.calendar_page_data() # app.init_data() # app.create_chart() # app.create_page() # app.upload_ftp()