import os
from datetime import datetime, timedelta
import configparser
import logging
import json
import requests
from typing import Dict, Any
class RWorkInfo:
def __init__(self, project: str, webhook_url: str):
self.project_name = project
self.webhook_url = webhook_url
self.DEFAULT_EMPTY = "未提供" # 数据缺失时的默认显示文本
def _format_amount(self, amount: Any) -> str:
"""
金额格式化：转为万元单位（保留2位小数），处理异常值
:param amount: 原始金额（整数/浮点数/None）
:return: 格式化后的金额字符串（如“180.56万”）
"""
try:
if amount is None:
return self.DEFAULT_EMPTY
amount_float = float(amount)
return f"{amount_float / 10000:.2f}万"
except (ValueError, TypeError):
return self.DEFAULT_EMPTY
def _format_percent(self, ratio: Any) -> str:
"""
百分比格式化：保留1-2位小数，添加升降符号，处理异常值
:param ratio: 原始比例（小数/None，如-0.3947代表-39.47%）
:return: 格式化后的百分比字符串（如“-39.5%↓”）
"""
try:
if ratio is None:
return f"{self.DEFAULT_EMPTY}↓" # 缺失时默认带↓（与示例格式对齐）
ratio_float = float(ratio)
percent_str = f"{ratio_float * 100:.1f}%" # 保留1位小数（示例风格）
return f"{percent_str}↓" if ratio_float < 0 else f"{percent_str}↑"
except (ValueError, TypeError):
return f"{self.DEFAULT_EMPTY}↓"
def _format_customer_flow(self, flow: Any, is_cumulative: bool = False) -> str:
"""
客流格式化：普通客流保留整数，累计客流转“万人次”（保留2位小数）
:param flow: 原始客流（整数/浮点数/None）
:param is_cumulative: 是否为累计客流（是则转万人次）
:return: 格式化后的客流字符串（如“24973”或“86.21万人次”）
"""
try:
if flow is None:
return self.DEFAULT_EMPTY
flow_int = int(flow)
if is_cumulative:
return f"{flow_int / 10000:.2f}万人次"
return str(flow_int)
except (ValueError, TypeError):
return self.DEFAULT_EMPTY
def format_message(self, date: str, data: Dict[str, Any]) -> Dict[str, Any]:
logging.info(f"正在生成{self.project_name}{date}飞书日报卡片消息")
# 安全提取数据
weather_info=data.get("weather_info", {})
basic_info = data.get("mall_basic_info", {})
key_indices = data.get("mall_key_indices", {})
mem_data = data.get("mem_data", {})
top_stores = data.get("store_sale_top_10", [])
# 工具函数：格式化金额（万元）和百分比
def format_sales(amount: Any) -> str:
try:
return f"{float(amount)/10000:.2f}万"
except (ValueError, TypeError):
return "0.00万"
def format_percent_arrow(ratio: Any) -> str:
try:
val = float(ratio) * 100
return f"{val:.1f}%↓" if val < 0 else f"{val:.1f}%↑"
except (ValueError, TypeError):
return "0.0%"
def format_percent(ratio: Any) -> str:
try:
val = float(ratio) * 100
return f"{val:.1f}%"
except (ValueError, TypeError):
return "0.0%"
def format_number(value: Any) -> str:
try:
return f"{int(value)}"
except (ValueError, TypeError):
return "0"
# 构建销售前十店铺文本
store_items = []
for idx, store in enumerate(top_stores[:10], 1):
store_name = store.get("store_name", "未知店铺")
sales = format_sales(store.get("sales_amt", 0))
store_items.append({
"tag": "div",
"text": {
"tag": "plain_text",
"content": f"{idx}. {store_name}：{sales}"
}
})
# 构建卡片消息内容
elements = [
# 天气信息
{
"tag": "div",
"text": {
"tag": "plain_text",
"content": f"**天气情况**"
}
},
{
"tag": "div",
"fields": [
{
"is_short": False,
"text": {
"tag": "plain_text",
"content": f"天气: {weather_info.get('condition')} 最高气温：{weather_info.get('max_temp')}°C 最低气温：{weather_info.get('min_temp')}°C"
}
}
]
},
# 项目基础信息
{
"tag": "div",
"text": {
"tag": "plain_text",
"content": "**一、项目基础信息**"
}
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"GLA：{format_number(basic_info.get('gla', 0))}m²"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"面积开业率：{format_percent(basic_info.get('open_store_arearate', 0))}"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"在营店铺数量：{format_number(basic_info.get('act_store_cnt', 0))}家"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"全场店铺数量：{format_number(basic_info.get('all_store_cnt', 0))}家"
}
}
]
},
{"tag": "hr"}, # 分隔线
# 关键指标
{
"tag": "div",
"text": {
"tag": "plain_text",
"content": "**二、关键指标**"
}
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"当日销售额：{format_sales(key_indices.get('mall_daily_sales_amt', 0))}"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"周环比：{format_percent_arrow(key_indices.get('lastweek_daily_sales_circle_rate', 0))}"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"月累计零售：{format_sales(key_indices.get('month_sale_amt', 0))}"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"年累计零售：{format_sales(key_indices.get('year_sale_amt', 0))}"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"当日客流：{format_number(key_indices.get('cust_flow', 0))}人次"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"周环比：{format_percent_arrow(key_indices.get('lastweek_cust_circle_reate', 0))}"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"月累计客流：{format_sales(key_indices.get('month_cust_flow', 0))}"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"年累计客流：{format_sales(key_indices.get('year_cust_flow', 0))}"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"当日车流：{format_number(key_indices.get('car_flow', 0))}辆"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"周环比：{format_percent_arrow(key_indices.get('lastweek_car_circle_reate', 0))}"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"月累计车流：{format_sales(key_indices.get('month_car_flow', 0))}"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"年累计车流：{format_sales(key_indices.get('year_car_flow', 0))}"
}
}
]
},
{"tag": "hr"}, # 分隔线
# 会员数据模块
{
"tag": "div",
"text": {
"tag": "plain_text",
"content": "**三、会员数据**"
}
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"新增会员数：{format_number(mem_data.get('inc_mem_cnt', 0))}人"
}
},
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"消费会员数：{format_number(mem_data.get('mem_exp_cnt', 0))}人"
}
}
]
},
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "plain_text",
"content": f"会员消费占比：{format_percent(mem_data.get('mem_exp_ratio', 0))}"
}
}
]
},
{"tag": "hr"},
# 销售前十店铺模块
{
"tag": "div",
"text": {
"tag": "plain_text",
"content": "**四、销售前十店铺**"
}
},
*store_items, # 展开店铺列表
# 底部备注
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": "⚠️ 以上数据来源于万象生活总部数据平台"
}
}
]
# 返回符合飞书规范的消息结构
return {
"msg_type": "interactive",
"card": {
"config": {"wide_screen_mode": True},
"header": {
"title": {
"content": f"📊{date}{self.project_name}简报",
"tag": "plain_text"
},
"template": "green"
},
"elements": elements
}
}
def send_to_feishu(self, message: Dict[str, Any]) -> bool:
"""发送消息到飞书机器人"""
try:
headers = {'Content-Type': 'application/json'}
response = requests.post(
self.webhook_url,
headers=headers,
data=json.dumps(message)
)
if response.status_code == 200:
result = response.json()
if result.get('code') == 0:
return True
logging.error(f"发送失败: {result.get('msg')}")
return False
logging.error(f"HTTP错误: {response.status_code}")
return False
except Exception as e:
logging.exception(f"发送消息时发生错误: {str(e)}")
return False
# Configure logging system
class LoggerSetup:
@staticmethod
def setup_logger():
"""设置按日期分割的日志记录器"""
# 创建logs目录
script_dir = os.path.dirname(os.path.abspath(__file__))
log_dir = os.path.join(script_dir, 'logs')
os.makedirs(log_dir, exist_ok=True)
# 创建日期对应的日志文件
log_file = os.path.join(
log_dir,
f"dailysalescheck_{datetime.now().strftime('%Y-%m-%d')}.log"
)
# 配置日志格式
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler(log_file, encoding='utf-8'),
logging.StreamHandler()
]
)
class FeishuWeatherBot:
def get_weather(self, city):
"""
获取天气信息
"""
weather_api_key = "天气平台API"
weather_api_url = "https://api.weatherapi.com/v1/forecast.json"
params = {
"key": weather_api_key,
"q": city,
"days": 3, # 修改为获取3天的预报
"aqi": "yes",
"lang": "zh"
}
try:
response = requests.get(weather_api_url, params=params)
weather_data = response.json()
return weather_data
except Exception as e:
print(f"获取天气信息失败: {str(e)}")
return None
def get_historical_weather(self, city, start_date, end_date=None):
"""
获取历史天气信息
:param city: 城市名
:param start_date: 起始日期 (格式: YYYY-MM-DD)
:param end_date: 结束日期 (格式: YYYY-MM-DD, 可选)
:return: 历史天气数据或 None
"""
weather_api_key = "fa8eab70fd174443aa873221251104"
weather_api_url = "https://api.weatherapi.com/v1/history.json"
params = {
"key": weather_api_key,
"q": city,
"dt": start_date, # 指定查询日期
"lang": "zh"
}
if end_date:
params["end_dt"] = end_date # 可选结束日期
try:
response = requests.get(weather_api_url, params=params)
weather_data = response.json()
return weather_data
except Exception as e:
print(f"获取历史天气信息失败: {str(e)}")
return None
class DataPlatform:
def __init__(self, base_url: str, app_id: str, app_secret: str):
self.base_url = base_url
self.app_id = app_id
self.app_secret = app_secret
self.session = requests.Session()
self.authorization = self.get_authorization()
def get_authorization(self) -> str:
"""获取认证token"""
headers = {
'Accept': '*/*',
'Content-Type': 'application/json',
}
json_data = {
'appId': self.app_id,
'appSecret': self.app_secret
}
response = self.session.post(f'{self.base_url}/login/appLogin',
headers=headers, json=json_data).json()
return response['data']
def get_api_data_exclude_params(self, api_id: str) -> Dict[str, Any]:
"""获取API数据（不包含参数）"""
headers = {
'Accept': '*/*',
'Content-Type': 'application/json',
'Authorization': self.authorization
}
json_data = {'apiId': api_id}
response = self.session.post(f'{self.base_url}/callApi/callApiById',
headers=headers, json=json_data).json()
return response
def get_api_data_include_params(self, api_id: str, params: Dict[str, str]) -> Dict[str, Any]:
"""带参数的API调用"""
headers = {
'Accept': '*/*',
'Authorization': self.authorization,
'Content-Type': 'application/json'
}
json_data = {
'apiId': api_id,
'params': params
}
response = self.session.post(f'{self.base_url}/callApi/callApiById',
headers=headers, json=json_data).json()
return response
def calculate_total_sales(data):
"""
计算销售数据列表中所有门店的销售额总和
参数:
data: 包含销售数据的列表，每个元素是一个字典，需包含'sales_amt'键
返回:
total: 所有门店的销售额总和
"""
total = 0.0
for item in data:
# 确保'sales_amt'存在且不为None
if 'sales_amt' in item and item['sales_amt'] is not None:
total += item['sales_amt']
return total
def main(app_id,app_secret,base_url,projectid):
# 创建平台实例
platform = DataPlatform(base_url, app_id, app_secret)
#获取当前日期
current_date = datetime.now()
logging.info(f"当前日期：{current_date}")
#获取昨天日期
last_date = current_date - timedelta(days=1)
last_date = last_date.strftime("%Y-%m-%d")
#获取上周同一天日期
last_week_date = current_date - timedelta(days=8)
last_week_date = last_week_date.strftime("%Y-%m-%d")
#创建天气实例
weather=FeishuWeatherBot()
#城市是北京
city = "Beijing"
# 获取北京昨日的天气
weather_data = weather.get_historical_weather(city, last_date)
# 初始化 weather_info with default values
if not weather_data:
max_temp = "未知"
min_temp = "未知"
condition = "未知"
else:
try:
for day in weather_data["forecast"]["forecastday"]:
max_temp = day["day"]["maxtemp_c"]
min_temp = day["day"]["mintemp_c"]
condition = day["day"]["condition"]["text"]
break # Only take the first day's data
except (KeyError, TypeError, IndexError):
# Fallback to default values if any error occurs during parsing
max_temp = "未知"
min_temp = "未知"
condition = "未知"
#构建天气数据字典
weather_info={
"day_code":last_date,
"max_temp":max_temp,
"min_temp":min_temp,
"condition":condition
}
params={
"day_code":last_date
}
# 获取店铺面积信息
try:
store_area = platform.get_api_data_include_params("merchant_org_rent_pick_open_story_area_for_any_region", params)
except Exception as e:
logging.error("接口调用失败: merchant_org_rent_pick_open_story_area_for_any_region, 错误: %s", str(e))
store_area = {'data': []} # 设置默认空值以防止后续索引错误
try:
store_area_info = [item for item in store_area.get('data', [])][0]
#项目GLA、店铺面积、项目开店面积比率
gla=store_area_info.get('gla')
act_store_area=store_area_info.get('act_store_area')
open_store_arearate=act_store_area/gla
except IndexError:
logging.warning("未获取到店铺面积信息数据，data为空")
store_area_info = {}
# 获取店铺数量信息
try:
store_count = platform.get_api_data_include_params("merchant_org_rent_pick_open_story_nums_for_any_region", params)
except Exception as e:
logging.error("接口调用失败: merchant_org_rent_pick_open_story_nums_for_any_region, 错误: %s", str(e))
store_count = {'data': []} # 设置默认空值以防止后续索引错误
try:
store_count_info = [item for item in store_count.get('data', [])][0]
#项目已开业店铺数量、项目总店铺数量
act_store_cnt=store_count_info.get('act_store_cnt')
all_store_cnt=store_count_info.get('act_store_cnt')+store_count_info.get('acpt_peno_store_cnt')+store_count_info.get('sigd_entd_store_cnt')+store_count_info.get('un_sigd_store_cnt')
except IndexError:
logging.warning("未获取到店铺数量信息数据，data为空")
store_count_info = {}
#项目基础信息
mall_basic_info = {
"gla": gla,
"act_store_area": act_store_area,
"open_store_arearate": open_store_arearate,
"act_store_cnt": act_store_cnt,
"all_store_cnt": all_store_cnt,
}
#print(mall_basic_info)
#获取店铺销售数据参数
mall_daily_sales_params={
"day_date_end":last_date,
"day_date_start":last_date,
"org_code_list":[projectid],
}
# 获取店铺销售数据
try:
mall_daily_sales = platform.get_api_data_include_params("org_sales_day_sum_for_any_region", mall_daily_sales_params)
except Exception as e:
logging.error("接口调用失败: org_sales_day_sum_for_any_region, 错误: %s", str(e))
mall_daily_sales = {'data': []} # 设置默认空值以防止后续索引错误
try:
#昨日商场销售金额
mall_daily_sales_info = [item for item in mall_daily_sales.get('data', [])][0]
mall_daily_sales_amt:float = mall_daily_sales_info.get('sales_amt')
except IndexError:
logging.warning("未获取到店铺销售数据，data为空")
mall_daily_sales_info = {}
# 查询上周同一天参数
mall_lastweek_daily_sales_params = {
"day_date_end": last_week_date,
"day_date_start": last_week_date,
"org_code_list": [projectid],
}
# 获取上周同一天商场销售数据
try:
mall_lastweek_daily_sales = platform.get_api_data_include_params("org_sales_day_sum_for_any_region", mall_lastweek_daily_sales_params)
except Exception as e:
logging.error("接口调用失败: org_sales_day_sum_for_any_region (上周数据), 错误: %s", str(e))
mall_lastweek_daily_sales = {'data': []} # 设置默认空值以防止后续索引错误
try:
mall_lastweek_daily_sales_info = [item for item in mall_lastweek_daily_sales.get('data', [])][0]
#上周同一天商场销售金额
mall_lastweek_daily_sales_amt:float = mall_lastweek_daily_sales_info.get('sales_amt')
#商场日销售金额周环比
lastweek_daily_sales_circle_rate=(mall_daily_sales_amt-mall_lastweek_daily_sales_amt)/mall_lastweek_daily_sales_amt
except IndexError:
logging.warning("未获取到上周商场销售数据，data为空")
mall_lastweek_daily_sales_info = {}
# 获取客车流数据参数
mall_daily_custcar_params = {
"date_day_start": last_date,
"date_day_end": last_date,
"org_code_list": [projectid],
}
# 获取客车流数据
try:
mall_cust_car_flow = platform.get_api_data_include_params("org_cust_car_flow_for_any_region", mall_daily_custcar_params)
except Exception as e:
logging.error("接口调用失败: org_cust_car_flow_for_any_region, 错误: %s", str(e))
mall_cust_car_flow = {'data': []} # 设置默认空值以防止后续索引错误
try:
mall_cust_car_flow_info = [item for item in mall_cust_car_flow.get('data', [])][0]
#昨日商场客户车辆数
cust_flow=int(mall_cust_car_flow_info['cust_flow'])
car_flow=int(mall_cust_car_flow_info['car_flow'])
except IndexError:
logging.warning("未获取到客车流数据，data为空")
mall_cust_car_flow_info = {}
# 查询上周同一天客车流参数
mall_lastweek_daily_custcar_params = {
"date_day_start": last_week_date,
"date_day_end": last_week_date,
"org_code_list": [projectid],
}
# 获取上周同一天客车流数据
try:
mall_lastweek_daily_custcar_flow = platform.get_api_data_include_params(
"org_cust_car_flow_for_any_region",
mall_lastweek_daily_custcar_params
)
except Exception as e:
logging.error("接口调用失败: org_cust_car_flow_for_any_region (上周数据), 错误: %s", str(e))
mall_lastweek_daily_custcar_flow = {'data': []} # 默认空值防止后续索引错误
try:
mall_lastweek_daily_custcar_flow_info = [item for item in mall_lastweek_daily_custcar_flow.get('data', [])][0]
except IndexError:
logging.warning("未获取到上周客车流数据，data为空")
mall_lastweek_daily_custcar_flow_info = {}
# 上周同一天商场客车流数
try:
lastweek_cust_flow = int(mall_lastweek_daily_custcar_flow_info['cust_flow'])
lastweek_car_flow = int(mall_lastweek_daily_custcar_flow_info['car_flow'])
except KeyError:
logging.warning("上周客车流数据中缺少 cust_flow 或 car_flow 字段")
lastweek_cust_flow = 0
lastweek_car_flow = 0
# 客户流量周环比计算（带除零保护）
if lastweek_cust_flow != 0:
lastweek_cust_circle_reate = (cust_flow - lastweek_cust_flow) / lastweek_cust_flow
else:
logging.warning("上周客户流量为零，无法计算周环比")
lastweek_cust_circle_reate = 0.0
# 车辆流量周环比计算（带除零保护）
if lastweek_car_flow != 0:
lastweek_car_circle_reate = (car_flow - lastweek_car_flow) / lastweek_car_flow
else:
logging.warning("上周车辆流量为零，无法计算周环比")
lastweek_car_circle_reate = 0.0
#print(lastweek_cust_flow,lastweek_car_flow,lastweek_cust_circle_reate,lastweek_car_circle_reate)
# 获取项目月累计和年累计参数
project_month_total_params = {
"date": last_date,
"org_code_list": [projectid],
}
# 获取项目月累计和年累计数据
try:
project_month_total = platform.get_api_data_include_params(
"day_org_weekly-monthly-year_performance_for_any_region",
project_month_total_params
)
except Exception as e:
logging.error("接口调用失败: day_org_weekly-monthly-year_performance_for_any_region, 错误: %s", str(e))
project_month_total = {'data': []} # 默认空值防止后续索引错误
try:
project_month_total_info = [item for item in project_month_total.get('data', [])][0]
#年累计销售额
year_sale_amt=project_month_total_info.get('year_sale_amt')
#月累计销售额
month_sale_amt=project_month_total_info.get('month_sale_amt')
#年累计车流辆数
year_car_flow=int(project_month_total_info['year_car_flow'])
#月累计车流车辆数
month_car_flow=int(project_month_total_info['month_car_flow'])
#月累计客流数
month_cust_flow=int(project_month_total_info['month_cust_flow'])
#年累计客流数
year_cust_flow=int(project_month_total_info['year_cust_flow'])
except IndexError:
logging.warning("未获取到项目月/年累计数据，data为空")
project_month_total_info = {}
#关键指标数据（不含车）
mall_key_indices_dict={
"mall_daily_sales_amt":mall_daily_sales_amt,
"lastweek_daily_sales_circle_rate":lastweek_daily_sales_circle_rate,
"cust_flow":cust_flow,
"lastweek_cust_circle_reate":lastweek_cust_circle_reate,
"car_flow":car_flow,
"lastweek_car_circle_reate":lastweek_car_circle_reate,
"year_sale_amt":year_sale_amt,
"month_sale_amt":month_sale_amt,
"year_car_flow":year_car_flow,
"month_car_flow":month_car_flow,
"month_cust_flow":month_cust_flow,
"year_cust_flow":year_cust_flow,
}
# 获取会员数据参数
member_params = {
"day_code": last_date,
"org_code_list": [projectid],
}
# 获取会员数据
try:
member_data = platform.get_api_data_include_params("member_consume_day_stats_for_any_region", member_params)
except Exception as e:
logging.error("接口调用失败: member_consume_day_stats_for_any_region, 错误: %s", str(e))
member_data = {'data': []} # 默认空值防止后续索引错误
try:
member_data_info = [item for item in member_data.get('data', [])][0]
#新增会员数
inc_mem_cnt=member_data_info.get('inc_mem_cnt')
#消费会员数
mem_exp_cnt=member_data_info.get('mem_exp_cnt')
#会员消费占比
mem_exp_ratio=member_data_info.get('mem_exp_ratio')
except IndexError:
logging.warning("未获取到会员数据，data为空")
member_data_info = {}
#print(inc_mem_cnt,mem_exp_cnt,mem_exp_ratio)
#会员数据
mem_data_dict={
"inc_mem_cnt":inc_mem_cnt,
"mem_exp_cnt":mem_exp_cnt,
"mem_exp_ratio":mem_exp_ratio,
}
# 获取店铺销售参数
store_sale_params = {
"day_date_end": last_date,
"day_date_start": last_date,
"org_code_list": [projectid],
}
# 获取店铺销售数据
try:
store_sale = platform.get_api_data_include_params("store_sales_day_sum_for_any_region", store_sale_params)
except Exception as e:
logging.error("接口调用失败: store_sales_day_sum_for_any_region, 错误: %s", str(e))
store_sale = {'data': []} # 默认空值防止后续索引错误
# 提取店铺销售数据
try:
store_sale_data_info = [item for item in store_sale.get('data', [])]
store_sale_sort_data=sorted(store_sale_data_info,key=lambda x:x['sales_amt'] if x['sales_amt'] is not None else -1,reverse=True)
store_sale_top_10=store_sale_sort_data[:10]
except IndexError:
logging.warning("未获取到店铺销售数据，data为空")
store_sale_data_info = []
#销售前十店铺
#print(store_sale_top_10)
mall_daily_sales_brief_data={
"weather_info":weather_info,
"mall_basic_info":mall_basic_info,
"mall_key_indices":mall_key_indices_dict,
"mem_data":mem_data_dict,
"store_sale_top_10":store_sale_top_10,
}
return mall_daily_sales_brief_data
if __name__ == '__main__':
# 初始化日志系统
LoggerSetup.setup_logger()
config = configparser.ConfigParser()
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.ini')
print(config_path)
if os.path.exists(config_path):
config.read(config_path, encoding='utf-8')
logging.info(f"配置文件已加载: {config_path}")
else:
# 创建默认配置文件
config['interface'] = {
'appId': '这里填写appId',
'appSecret': '这里填写appSecret',
}
config['rworkinfo'] = {
'webhook_url': '这里填写飞书机器人WebHook URL',
'projectid': '这里填写项目ID',
'project': '这里填写项目名称',
'excludestore': '这里填写需要排除的店铺编号，多个店铺用英文逗号分隔'
}
with open(config_path, 'w') as f:
config.write(f)
logging.info(f"已创建默认配置文件: {config_path}")
# 读取配置
app_id = config.get('interface', 'appId')
app_secret = config.get('interface', 'appSecret')
base_url = "数据平台接口地址"
webhook_url = config.get('rworkinfo', 'webhook_url')
projectid = config.get('rworkinfo', 'projectid')
project = config.get('rworkinfo', 'project')
mall_daily_sales_brief_data=main(app_id,app_secret,base_url,projectid)
#print(mall_daily_sales_brief_data)
#创建RWork对象
rwork=RWorkInfo(project,webhook_url)
last_date = datetime.now() - timedelta(days=1)
last_date = last_date.strftime("%Y-%m-%d")
#格式化模板参数
messages=rwork.format_message(last_date,mall_daily_sales_brief_data)
#print(json.dumps(messages, ensure_ascii=False, indent=2))
#发送消息
logging.info("开始发送消息")
try:
rwork.send_to_feishu(messages)
logging.info("消息已成功发送至*工作")
except Exception as e:
logging.error("发送消息至*工作失败: %s", str(e))