diff options
Diffstat (limited to 'community')
| -rw-r--r-- | community/disc.py | 28 | ||||
| -rw-r--r-- | community/wacca_plus/wacca_plus.py | 132 |
2 files changed, 160 insertions, 0 deletions
diff --git a/community/disc.py b/community/disc.py new file mode 100644 index 0000000..8cc5aeb --- /dev/null +++ b/community/disc.py @@ -0,0 +1,28 @@ +import os +import requests +from dotenv import load_dotenv + +load_dotenv() + +def fetch_messages(channel_id: str): + url = f"https://discord.com/api/v9/channels/{channel_id}/messages?limit=50" + headers = { + "accept-encoding": "gzip, deflate, br, zstd", + "accept-language": "en-GB", + "authorization": os.getenv("DISCORD_AUTHORIZATION"), # Replace with your real token + "priority": "u=1, i", + "sec-ch-ua": '"Not:A-Brand";v="24", "Chromium";v="134"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Linux"', + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-origin", + "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36", + "x-debug-options": "bugReporterEnabled", + "x-discord-locale": "en-US", + "x-discord-timezone": "America/Vancouver", + } + response = requests.get(url, headers=headers) + return response.json() + if response.ok: + print(f"Failed to fetch: {response.status_code} - {response.text}") diff --git a/community/wacca_plus/wacca_plus.py b/community/wacca_plus/wacca_plus.py new file mode 100644 index 0000000..023a7ae --- /dev/null +++ b/community/wacca_plus/wacca_plus.py @@ -0,0 +1,132 @@ +import os +from datetime import datetime +import time +import requests +import openai +import json +from dotenv import load_dotenv +import base64 + +load_dotenv() + +def check_is_generation_possible(): + return os.getenv("OPENAI_API_KEY") is not None and os.getenv("DISCORD_AUTHORIZATION") is not None + + +def check_is_announcement_image(img_url: str): + openai.api_key = os.getenv("OPENAI_API_KEY") + + tools = [ + { + "type": "function", + "function": { + "name": "classify_wacca_plus_image", + "description": "Classify if an image is WACCA PLUS announcement, update, or information", + "parameters": { + "type": "object", + "properties": { + "is_wacca_plus_related": { + "type": "boolean", + "description": "Is this image related to WACCA PLUS?", + }, + "category": { + "type": "string", + "enum": ["announcement", "update", "info", "null"], + "description": "Category of image if related; otherwise null.", + }, + }, + "required": ["is_wacca_plus_related", "category"], + }, + } + } + ] + + response = openai.chat.completions.create( + model="gpt-4o", + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "Does this image contain official update, event, or announcement information for the game WACCA PLUS? Ignore unrelated content like gameplay screenshots, score posts, or arcade cabinet photos. Classify accordingly."}, + {"type": "image_url", "image_url": {"url": img_url}}, + ], + } + ], + tools=tools, + tool_choice={"type": "function", "function": {"name": "classify_wacca_plus_image"}}, + ) + + tool_args = response.choices[0].message.tool_calls[0].function.arguments + parsed_result = json.loads(tool_args) + return parsed_result["is_wacca_plus_related"], parsed_result["category"] + +def _load_cache(): + cache_file = "wac_result_cache.json" + if not os.path.exists(cache_file): + with open(cache_file, "w") as file: + json.dump({}, file) + with open(cache_file, "r") as file: + return json.load(file) + +def _save_cache(cache: dict): + cache_file = "wac_result_cache.json" + with open(cache_file, "w") as file: + json.dump(cache, file) + +def _convert_image_to_base64(img_url: str): + response = requests.get(img_url) + if response.status_code == 200: + img_data = response.content + img_base64 = base64.b64encode(img_data).decode('utf-8') + mime_type = response.headers['Content-Type'] + return f"data:{mime_type};base64,{img_base64}" + else: + raise Exception(f"Failed to fetch image from URL: {img_url}, status code: {response.status_code}") + +def parse_announcement_messages(message_json: dict): + news_posts = [] + cache = _load_cache() + for message in message_json: + type = None + if len(message["attachments"]) == 0: + continue + image_attachments = [] + for attachment in message["attachments"]: + if "image" in attachment["content_type"]: + image_attachments.append(attachment) + + if len(image_attachments) == 0: + continue + + filtered_images = [] + for image in image_attachments: + if image["id"] in cache: + is_related = cache[image["id"]][0] + type = cache[image["id"]][1] + else: + is_related, type = check_is_announcement_image(image["url"]) + cache[image["id"]] = [is_related, type] + if not is_related: + continue + filtered_images.append({"image": _convert_image_to_base64(image["url"]), "url": None}) + + if len(filtered_images) == 0: + continue + + date = message["timestamp"].split("T")[0] + date_obj = datetime.strptime(date, "%Y-%m-%d") + unix_time = int(time.mktime(date_obj.timetuple())) + + news_posts.append({ + "date": date, + "identifier": "WACCA_PLUS", + "type": type, + "timestamp": unix_time, + "content": "NEW INFORMATION FROM WACCA+ / WACCA+ の最新情報", + "headline": None, + "url": None, + "images": filtered_images + }) + + _save_cache(cache) + return news_posts |