Spaces:
Running
Running
File size: 3,214 Bytes
6900d0c 999b610 6900d0c a0196ab 6900d0c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
import psycopg2
import datetime
from bin_public.config.presets import *
from dateutil import tz
import os
def current_time(type):
if type == 'ymd':
return datetime.datetime.now().strftime("%Y-%m-%d")
if type == 'ymdhms':
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# hologres 基础函数：查询
def holo_query_func(run_sql, is_query=0):
conn = psycopg2.connect(host=os.environ['HOST'],
port=os.environ['PORT'],
dbname=os.environ['DBNAME'],
user=os.environ['AK'],
password=os.environ['SK'])
cur = conn.cursor()
cur.execute(run_sql)
if is_query:
data = cur.fetchall()
cur.close()
conn.close()
if is_query:
return data
def holo_query_account_mapping(invite_code):
run_sql = f"""
select end_date, status, mapping_ak
from s_account_invite_code
where invite_code = '{invite_code}'
order by gmt_modify desc
limit 1
"""
data = holo_query_func(run_sql, is_query=1)
# 数据库中查不到，则返回no_invite_code_msg
if len(data) == 0:
status_text = standard_error_msg + no_invite_code_msg
return status_text, None
# 数据库中查到，判断是否可用
if len(data) == 1:
end_date = data[0][0]
status = data[0][1]
mapping_ak = data[0][2]
if end_date < datetime.datetime.now().strftime("%Y%m%d") or status != '1':
status_text = standard_error_msg + no_useful_invite_code_msg
return status_text, None
return 'Success status: ready', mapping_ak
def key_preprocessing(keyTxt):
invite_code = keyTxt
# 这里先用这个逻辑，到时候等实际的邀请码来了就改一下这个函数就行
if keyTxt.startswith("dteam_"):
status_display, keyTxt = holo_query_account_mapping(keyTxt)
yield status_display, keyTxt, invite_code
return
else:
if len(keyTxt) != 51:
status_display = standard_error_msg + no_apikey_msg
yield status_display, keyTxt, invite_code
return
yield 'Success status: ready', keyTxt, invite_code
return
def holo_query_insert_chat_message(invite_code, prompt, response, all_token_cnt, history):
run_sql = f"""
insert into s_account_chat_message(
gmt_create
,invite_code
,prompt
,response
,all_token_cnt
,history
,chat_seq
,log_timestamp
)
select
'{datetime.datetime.now().replace(tzinfo=tz.gettz('Asina/Shanghai')).strftime("%Y-%m-%d %H:%M:%S")}' as gmt_create
,'{str(invite_code).replace("'", '"')}' as invite_code
,'{str(prompt).replace("'", '"')}' as prompt
,'{str(response).replace("'", '"')}' as response
,'{str(all_token_cnt).replace("'", '"')}' as all_token_cnt
,'{str(history).replace("'", '"')}' as history
,'{len(history)}' as chat_seq
,localtimestamp as log_timestamp
"""
holo_query_func(run_sql, is_query=0) |