(chat_id: int, record_id: int, chart: dict, data: dict)
| 1689 | |
| 1690 | |
| 1691 | def request_picture(chat_id: int, record_id: int, chart: dict, data: dict): |
| 1692 | file_name = f'c_{chat_id}_r_{record_id}' |
| 1693 | |
| 1694 | columns = chart.get('columns') if chart.get('columns') else [] |
| 1695 | x = None |
| 1696 | y = None |
| 1697 | series = None |
| 1698 | multi_quota_fields = [] |
| 1699 | multi_quota_name = None |
| 1700 | |
| 1701 | if chart.get('axis'): |
| 1702 | axis_data = chart.get('axis') |
| 1703 | x = axis_data.get('x') |
| 1704 | y = axis_data.get('y') |
| 1705 | series = axis_data.get('series') |
| 1706 | # 获取multi-quota字段列表 |
| 1707 | if axis_data.get('multi-quota') and 'value' in axis_data.get('multi-quota'): |
| 1708 | multi_quota_fields = axis_data.get('multi-quota').get('value', []) |
| 1709 | multi_quota_name = axis_data.get('multi-quota').get('name') |
| 1710 | |
| 1711 | axis = [] |
| 1712 | for v in columns: |
| 1713 | axis.append({'name': v.get('name'), 'value': v.get('value')}) |
| 1714 | if x: |
| 1715 | axis.append({'name': x.get('name'), 'value': x.get('value'), 'type': 'x'}) |
| 1716 | if y: |
| 1717 | y_list = y if isinstance(y, list) else [y] |
| 1718 | |
| 1719 | for y_item in y_list: |
| 1720 | if isinstance(y_item, dict) and 'value' in y_item: |
| 1721 | y_obj = { |
| 1722 | 'name': y_item.get('name'), |
| 1723 | 'value': y_item.get('value'), |
| 1724 | 'type': 'y' |
| 1725 | } |
| 1726 | # 如果是multi-quota字段，添加标志 |
| 1727 | if y_item.get('value') in multi_quota_fields: |
| 1728 | y_obj['multi-quota'] = True |
| 1729 | axis.append(y_obj) |
| 1730 | if series: |
| 1731 | axis.append({'name': series.get('name'), 'value': series.get('value'), 'type': 'series'}) |
| 1732 | if multi_quota_name: |
| 1733 | axis.append({'name': multi_quota_name, 'value': multi_quota_name, 'type': 'other-info'}) |
| 1734 | |
| 1735 | request_obj = { |
| 1736 | "path": os.path.join(settings.MCP_IMAGE_PATH, file_name), |
| 1737 | "type": chart.get('type'), |
| 1738 | "data": orjson.dumps(data.get('data') if data.get('data') else []).decode(), |
| 1739 | "axis": orjson.dumps(axis).decode(), |
| 1740 | } |
| 1741 | |
| 1742 | _error = None |
| 1743 | try: |
| 1744 | requests.post(url=settings.MCP_IMAGE_HOST, json=request_obj, timeout=settings.SERVER_IMAGE_TIMEOUT) |
| 1745 | except Exception as e: |
| 1746 | _error = e |
| 1747 | |
| 1748 | request_path = urllib.parse.urljoin(settings.SERVER_IMAGE_HOST, f"{file_name}.png") |
no test coverage detected