import math import requests import json from django.conf import settings from django.utils import timezone from celery.utils.log import get_task_logger from pingcap_sso.celery import app as celery_app from pingcap_sso.discourse.models import ( DiscourseTopicCustomField, ) from pingcap_sso.discourse.consts import QA_FEEDBACK_CATEGORIES from pingcap_sso.models import ( UserTiDBUsage, User ) from pingcap_sso.discourse.models import ( DiscourseTopicCustomField, DiscourseTopic, DiscoursePost, ) from pingcap_sso.discourse.lib import extract_tidb_version_from_topic logger = get_task_logger(__name__) @celery_app.task def update_user_tidb_usage_by_asktug_qa_topic(topic_id): tidb_usage, created = DiscourseTopicCustomField.objects.get_or_create( topic_id=topic_id, name='version', ) if tidb_usage.value: return try: topic = DiscourseTopic.objects.get( id=topic_id, category_id__in=QA_FEEDBACK_CATEGORIES, deleted_at__isnull=True, archetype='regular', visible=1, closed=0 ) except DiscourseTopic.DoesNotExist: return tidb_version = extract_tidb_version_from_topic(topic) if not tidb_version: return tidb_usage.value = f'v{ tidb_version }' tidb_usage.save() @celery_app.task def sync_asktug_topic_version_to_tidb_usage(): asktug_version_qs = DiscourseTopicCustomField.objects.select_related( 'topic', 'topic__user' ).exclude(value='').exclude(value__isnull=True).filter( name='version', topic__category_id__in=QA_FEEDBACK_CATEGORIES, ) count = asktug_version_qs.count() step = 500 for i in range(math.ceil(count / step)): asktug_topics = list( asktug_version_qs.values_list( 'topic_id', 'value', 'topic__user__username', 'topic__created_at' )[i * step: (i + 1) * step] ) topic_ids = [v[0] for v in asktug_topics] tidb_usages = dict( UserTiDBUsage.objects.filter( asktug_topic_id__in=topic_ids ).values_list( 'asktug_topic_id', 'version' ) ) for topic_id, asktug_version, username, topic_created_at in asktug_topics: if asktug_version: # asktug store tidb version with prefix `v` asktug_version = asktug_version[1:] else: continue account_version = tidb_usages.get(topic_id) if account_version: # already synced, check if the version is the same if account_version.lower() == asktug_version.lower(): continue else: UserTiDBUsage.objects.filter( asktug_topic_id=topic_id ).update( version=asktug_version ) else: try: user = User.objects.get(username=username) except User.DoesNotExist: logger.error(f'User { username } not exists') continue UserTiDBUsage.objects.create( user=user, asktug_topic_id=topic_id, version=asktug_version, topic_created_at=topic_created_at, ) @celery_app.task def sync_asktug_topic_deleted_at_to_tidb_usage(): step = 500 usage_qs = UserTiDBUsage.objects.all().order_by('id') count = usage_qs.count() for i in range(math.ceil(count / step)): _usage_qs = usage_qs[i * step: (i + 1) * step] topic_ids = [usage.asktug_topic_id for usage in _usage_qs] topic_deleted_at_dict = dict( DiscourseTopic.objects.filter(id__in=topic_ids).values_list( 'id', 'deleted_at' ) ) for usage in _usage_qs: if usage.asktug_topic_id not in topic_deleted_at_dict: continue new_deleted_at = topic_deleted_at_dict[usage.asktug_topic_id] if usage.topic_deleted_at != new_deleted_at: usage.topic_deleted_at = new_deleted_at usage.save(update_fields=['topic_deleted_at']) TOPIC_USAGE_EXTRACTION_PROMPT = """ 你是 TiDB 数据库社区论坛的助理机器人，任务是从用户发送的技术提问帖子中提取运营需要的关键信息，如果你没有在帖子中识别到对应信息那么留空就行, 你可以根据用户的描述进行合理推断, 但不可以毫无依据的编造数据。最后以 json 格式输出结果。 你需要提取一下信息: - 字段: enviroment 说明: 当前在什么环境使用 tidb 选项: "production", "test", "poc" - 字段: cloud_provider 说明: - TiDB 部署在哪个云厂商，比如 "aws", "gcp", "azure", "ucloud", "阿里云", "腾讯云", "金山云" 之类 - 你可以根据用户的描述进行合理推断, 比如当用户虽然没有明确提到云厂商名称, 但是提到了使用了某个云厂商的组件, 那么可以推断用户正在使用或者准备使用的是该云厂商的云服务 - 用户如果是打算使用云厂商在进行调研，那也算作是使用云厂商的云服务 - 如果是本地部署则留空 - 字段: deployment_opreation_system 说明: 部署 TiDB 使用的机器操作系统，例如 "ubuntu", "centos" - 字段: deployment_type 说明: 部署方式，比如 "云上部署（什么机器配置、什么硬盘）", "本地部署(什么机器配置、什么硬盘)" 之类的 - 字段: tidb_cluster_data_size 说明: TiDB 集群数据量，包括： - 存储容量：如 "1TB", "100GB" - 数据行数：如 "单表215亿行", "1亿条记录", 如果有多个表，则用逗号隔开分开记录 - 表数量：如 "1000张表" **提取规则**： - 优先提取用户明确提到的具体数字 - 行数信息统一转换为"X亿行"或"X万行"格式 - 如果用户同时提到多种数据量信息，都提取出来用逗号分隔 - 字段: tidb_cluster_node_number 说明: TiDB 集群节点数，比如简略的 "10", "100" 或者详细的 "TiDB 10 节点,TiKV 10 节点,TiFlash 10 节点" **提取规则**： - 优先提取用户明确提到的具体数字 - 如果用户同时提到多种 TiDB 节点数信息，都提取出来用逗号分隔 ** 注意以下事项： - 绝对不要编造数据 - 社区论坛为用户提供了问题模板, 请严格识别用户实际填写的信息。如果用户保留了问题模板中的任何占位符，例如："云上部署（什么云）/机器部署（什么机器配置、什么硬盘）" 中的 "云上部署（什么云）" 或 "机器部署（什么机器配置、什么硬盘）" 等，即使只保留了前半部分，也应视作用户未在此提供有效信息。只有当用户明确填写了具体的部署方式，例如 "云上部署（阿里云）" 或 "本地部署（服务器型号XXX，SSD硬盘）" 时，才从这句话中提取有效信息。 - 输出请使用 JSON 格式 - 严格遵守下面的字段名称 { "enviroment": "test", "cloud_provider": "gcp", "deployment_opreation_system": "ubuntu", "deployment_type": "云上部署", "tidb_cluster_data_size": "1TB", "tidb_cluster_node_number": 100 } """ def extract_tidb_usage_from_text_by_deepseek(content): api_key = settings.DEEPSEEK_API_KEY response = requests.post( "https://api.deepseek.com/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json={ "model": "deepseek-chat", "messages": [ {"role": "system", "content": TOPIC_USAGE_EXTRACTION_PROMPT}, {"role": "user", "content": content}, ], "temperature": 1.0, "response_format": { 'type': 'json_object' } } ) if response.status_code != 200: raise Exception(f"请求失败: {response.status_code} {response.text}") result = json.loads(response.json()['choices'][0]['message']['content']) return result def extract_tidb_info_for_user_tidb_usage(usage_id): usage = UserTiDBUsage.objects.get(id=usage_id) if usage.ai_extracted_at: return try: post = DiscoursePost.objects.get(topic_id=usage.asktug_topic_id, post_number=1, deleted_at__isnull=True) except DiscoursePost.DoesNotExist: return False # extract only the first 2000 characters of the post, # as the key message is typically within this range. result = extract_tidb_usage_from_text_by_deepseek(post.raw[:2000]) usage.ai_extracted_at = timezone.now() usage.enviroment = (result.get('enviroment') or '')[:32] usage.cloud_provider = (result.get('cloud_provider') or '')[:32] usage.deployment_opreation_system = (result.get('deployment_opreation_system') or '')[:64] usage.deployment_type = (result.get('deployment_type') or '')[:64] usage.tidb_cluster_data_size = (result.get('tidb_cluster_data_size') or '')[:32] usage.tidb_cluster_node_number = str(result.get('tidb_cluster_node_number') or '')[:32] usage.save() @celery_app.task def bulk_extract_tidb_info_for_user_tidb_usage(): usage_qs = UserTiDBUsage.objects.filter(ai_extracted_at__isnull=True) for usage in usage_qs: extract_tidb_info_for_user_tidb_usage(usage.id)