Spaces:
Sleeping
Sleeping
| #pip install fastapi ###for fastapi | |
| #pip install uvicorn ###for server. to run the api serice from terminal: uvicorn main:app --reload | |
| #pip install gunicorn ###gunicorn --bind 0.0.0.0:8000 -k uvicorn.workers.UvicornWorker main:app | |
| #pip install python-multipart ###for UploadFile | |
| #pip install pillow ###for PIL | |
| #pip install transformers ###for transformers | |
| #pip install torch ###for torch | |
| #pip install sentencepiece ###for AutoTokenizer | |
| #pip install -U cos-python-sdk-v5 ###腾讯云对象存储SDK（COS-SDK） | |
| #pip install -q -U google-generativeai | |
| # from typing import Optional | |
| # from fastapi import FastAPI, Header | |
| # #from transformers import pipeline, EfficientNetImageProcessor, EfficientNetForImageClassification, AutoTokenizer, AutoModelForSeq2SeqLM | |
| # import torch | |
| # from transformers import EfficientNetImageProcessor, EfficientNetForImageClassification, pipeline | |
| # from models import ItemInHistory, ItemUploaded, ServiceLoginInfo | |
| # from openai import OpenAI | |
| # import sys, os, logging | |
| # import json, requests | |
| from fastapi import FastAPI | |
| import sys, logging, os | |
| from models import Item2AI, AI2Item | |
| import google.generativeai as genai | |
| import urllib.parse as urlparse | |
| from qcloud_cos import CosConfig, CosS3Client | |
| from PIL import Image | |
| # init app and logging | |
| app = FastAPI() | |
| logging.basicConfig(level=logging.INFO, stream=sys.stdout) | |
| # load google gemini models | |
| genai.configure(api_key=os.environ.get("GOOGLE_API_KEY")) | |
| gemini_pro = genai.GenerativeModel('gemini-pro') | |
| logging.info("google gemini-pro model loaded successfully.") | |
| gemini_pro_vision = genai.GenerativeModel('gemini-pro-vision') | |
| logging.info("google gemini-pro-vision model loaded successfully.") | |
| # init tencent cos | |
| cos_secret_id = os.environ['COS_SECRET_ID'] | |
| cos_secret_key = os.environ['COS_SECRET_KEY'] | |
| cos_region = 'ap-shanghai' | |
| cos_bucket = '7072-prod-3g52ms9o7a81f23c-1324125412' | |
| token = None | |
| scheme = 'https' | |
| config = CosConfig(Region=cos_region, SecretId=cos_secret_id, SecretKey=cos_secret_key, Token=token, Scheme=scheme) | |
| client = CosS3Client(config) | |
| logging.info(f"tencent cos init succeeded.") | |
| # Route to create an item | |
| async def ai_first_turn(item: Item2AI): | |
| logging.info("ai_first_turn...") | |
| logging.info("item:", item) | |
| # response = gemini_pro.generate_content("What is the meaning of life?") | |
| # logging.info(response.text) | |
| url = urlparse.urlparse(item.item_fileurl) | |
| key = url[2][1::] | |
| bucket = url[1].split('.')[1] | |
| contentfile = key.split('/')[1] | |
| historyid = contentfile.split('.')[0] | |
| response = client.get_object( | |
| Bucket = bucket, | |
| Key = key | |
| ) | |
| response['Body'].get_stream_to_file(contentfile) | |
| ai2item = AI2Item( | |
| upload_id = item.upload_id, | |
| union_id = item.union_id, | |
| item_fileurl = item.item_fileurl, | |
| item_mediatype = item.item_mediatype, | |
| upload_datetime = item.upload_datetime, | |
| ai_feedback = "" | |
| ) | |
| if item.item_mediatype == "image": | |
| # Opening the image using PIL | |
| img = Image.open(contentfile) | |
| logging.info(f"image file {contentfile} is opened.") | |
| response = gemini_pro_vision.generate_content(["Describe this picture in Chinese with plenty of details.", img]) | |
| ai2item.ai_feedback = response.text | |
| else: | |
| ai2item.ai_feedback = "不是image类型，暂不能识别" | |
| logging.info(ai2item) | |
| return(ai2item) | |
| # try: | |
| # ai_model_bc_preprocessor = EfficientNetImageProcessor.from_pretrained("./birds-classifier-efficientnetb2") | |
| # ai_model_bc_model = EfficientNetForImageClassification.from_pretrained("./birds-classifier-efficientnetb2") | |
| # logging.info(f"local model dennisjooo/Birds-Classifier-EfficientNetB2 loaded.") | |
| # except Exception as e: | |
| # logging.error(e) | |
| # try: | |
| # openai_client = OpenAI( | |
| # api_key=os.environ.get("OPENAI_API_KEY"), | |
| # ) | |
| # # prompt = """你是一个鸟类学家，用中文回答关于鸟类的问题。你的回答需要满足以下要求: | |
| # # 1. 你的回答必须是中文 | |
| # # 2. 回答限制在100个字以内""" | |
| # # conv = Conversation(open_client, prompt, 3) | |
| # logging.info(f"openai chat model loaded.") | |
| # except Exception as e: | |
| # logging.error(e) | |
| # try: | |
| # ai_model_bc_pipe= pipeline("image-classification", model="dennisjooo/Birds-Classifier-EfficientNetB2") | |
| # logging.info(f"remote model dennisjooo/Birds-Classifier-EfficientNetB2 loaded.") | |
| # except Exception as e: | |
| # print(e) | |
| #try: | |
| # ai_model_ez_preprocessor = AutoTokenizer.from_pretrained("./opus-mt-en-zh") | |
| # ai_model_ez_model = AutoModelForSeq2SeqLM.from_pretrained("./opus-mt-en-zh") | |
| # print(f"local model Helsinki-NLP/opus-mt-en-zh loaded.") | |
| #except Exception as e: | |
| # print(e) | |
| #try: | |
| # ai_model_ez_pipe= pipeline(task="translation_en_to_zh", model="Helsinki-NLP/opus-mt-en-zh", device=0) | |
| # print(f"remote model Helsinki-NLP/opus-mt-en-zh loaded.") | |
| #except Exception as e: | |
| # print(e) | |
| # def bird_classifier(image_file: str) -> str: | |
| # # Opening the image using PIL | |
| # img = Image.open(image_file) | |
| # logging.info(f"image file {image_file} is opened.") | |
| # result:str = "" | |
| # try: | |
| # inputs = ai_model_bc_preprocessor(img, return_tensors="pt") | |
| # # Running the inference | |
| # with torch.no_grad(): | |
| # logits = ai_model_bc_model(**inputs).logits | |
| # # Getting the predicted label | |
| # predicted_label = logits.argmax(-1).item() | |
| # result = ai_model_bc_model.config.id2label[predicted_label] | |
| # logging.info(f"{ai_model_bc_model.config.id2label[predicted_label]}:{ai_model_bc_pipe(img)[0]['label']}") | |
| # except Exception as e: | |
| # logging.error(e) | |
| # logging.info(result) | |
| # return result | |
| # def text_en_zh(text_en: str) -> str: | |
| # text_zh = "" | |
| # if ai_model_ez_status is MODEL_STATUS.LOCAL: | |
| # input = ai_model_ez_preprocessor(text_en) | |
| # translated = ai_model_ez_model.generate(**ai_model_ez_preprocessor(text_en, return_tensors="pt", padding=True)) | |
| # for t in translated: | |
| # text_zh += ai_model_ez_preprocessor.decode(t, skip_special_tokens=True) | |
| # elif ai_model_ez_status is MODEL_STATUS.REMOTE: | |
| # text_zh = ai_model_ez_pipe(text_en) | |
| # return text_zh | |
| # # Route to list all items uploaded by a specific user by unionid | |
| # # @app.get("/items/{user_unionid}") | |
| # # def list_items(user_unionid: str) -> dict[str, list[ItemInHistory]]: | |
| # # logging.info("list_items") | |
| # # logging.info(user_unionid) | |
| # # items: list[ItemInHistory] = [] | |
| # # response = client.list_objects( | |
| # # Bucket=cos_bucket, | |
| # # Prefix=f'{user_unionid}/history/' | |
| # # ) | |
| # # logging.info(response['Contents']) | |
| # # for obj in response['Contents']: | |
| # # key:str = obj['Key'] | |
| # # response = client.get_object( | |
| # # Bucket = cos_bucket, | |
| # # Key = key | |
| # # ) | |
| # # localfile = key.split('/')[2] | |
| # # response['Body'].get_stream_to_file(localfile) | |
| # # item = itemFromJsonFile(localfile) | |
| # # items.append(item) | |
| # return {"items": items} | |
| # # Route to list all items uploaded by a specific user by unionid from header | |
| # @app.get("/items/") | |
| # def list_items_byheader(x_wx_openid: Optional[str]=Header(None)) -> dict[str, list[ItemInHistory]]: | |
| # logging.info("list_items_byheader") | |
| # logging.info(x_wx_openid) | |
| # items: list[ItemInHistory] = [] | |
| # response = client.list_objects( | |
| # Bucket=cos_bucket, | |
| # Prefix=f'{x_wx_openid}/history/' | |
| # ) | |
| # logging.info(response['Contents']) | |
| # for obj in response['Contents']: | |
| # key:str = obj['Key'] | |
| # response = client.get_object( | |
| # Bucket = cos_bucket, | |
| # Key = key | |
| # ) | |
| # localfile = key.split('/')[2] | |
| # response['Body'].get_stream_to_file(localfile) | |
| # item = itemFromJsonFile(localfile) | |
| # items.append(item) | |
| # return {"items": items} | |
| # def itemFromJsonFile(jsonfile: str) -> ItemInHistory: | |
| # f = open(jsonfile, 'r') | |
| # content = f.read() | |
| # a = json.loads(content) | |
| # f.close() | |
| # return ItemInHistory(history_id = a['history_id'],union_id = a['union_id'], | |
| # item_fileurl = a['item_fileurl'],item_mediatype = a["item_mediatype"], | |
| # upload_datetime = a["upload_datetime"],ai_feedback = a['ai_feedback']) | |
| # def itemToJsonFile(item: ItemInHistory): | |
| # history_json = { | |
| # "history_id": item.history_id, | |
| # "union_id": item.union_id, | |
| # "item_fileurl": item.item_fileurl, | |
| # "item_mediatype": item.item_mediatype, | |
| # "upload_datetime": item.upload_datetime, | |
| # "ai_feedback": item.ai_feedback | |
| # } | |
| # b = json.dumps(history_json) | |
| # historyfile = f'{item.history_id}.json' | |
| # f = open(historyfile, 'w') | |
| # f.write(b) | |
| # f.close() | |
| # return historyfile | |