diff options
| author | A Farzat <a@farzat.xyz> | 2025-10-12 14:57:22 +0300 |
|---|---|---|
| committer | A Farzat <a@farzat.xyz> | 2025-10-12 14:57:22 +0300 |
| commit | e4d8ae3d88aac184ba7c15c3683630ea8e182e61 (patch) | |
| tree | 5dfd71824a4a33a96a73faeb8e3166aa887adcab /wsgi | |
| parent | 20c63cb6b18ee70b874ba63c1ad6212af165b291 (diff) | |
| download | csca5028-e4d8ae3d88aac184ba7c15c3683630ea8e182e61.tar.gz csca5028-e4d8ae3d88aac184ba7c15c3683630ea8e182e61.zip | |
Diffstat (limited to 'wsgi')
| -rw-r--r-- | wsgi/__init__.py | 97 | ||||
| -rw-r--r-- | wsgi/utils.py | 14 |
2 files changed, 0 insertions, 111 deletions
diff --git a/wsgi/__init__.py b/wsgi/__init__.py deleted file mode 100644 index 6224e6e..0000000 --- a/wsgi/__init__.py +++ /dev/null @@ -1,97 +0,0 @@ -from datetime import datetime, UTC -from typing import Any, Dict, List, Tuple - -from flask import Flask, request -from flask_cors import CORS -from pymongo.errors import DuplicateKeyError - -from components.database import subscriptions -from components.subscriptions.main import Subscription -from components.videos import VideoTuple -from components.extractor.extract_sub_info import get_sub_info_from_yt_url -from .utils import vid_dicts_from_tuple_list, sub_info_from_dict - -app = Flask(__name__) -CORS(app) - -@app.route("/vid-from-link/<id>") -def videos_from_link(id: str) -> Tuple[List[Dict[str, Any]], int]: - sub_dict = subscriptions.find_one({"_id": id}) - if sub_dict: - return vid_dicts_from_tuple_list(sub_dict["videos"]), 200 - return [{'error': "Subscription %s not found"%id }], 404 - -@app.route("/sub-info/<id>") -def sub_dict(id: str) -> Tuple[Dict[str, Any], int]: - sub_dict = subscriptions.find_one({"_id": id}) - if sub_dict: - return sub_info_from_dict(sub_dict), 200 - return {'error': "Subscription %s not found"%id }, 404 - -@app.route("/subs-info") -def subs_info() -> List[Dict[str, Any]]: - return [sub_info_from_dict(sub_dict) for sub_dict in subscriptions.find()] - -@app.post("/add-sub/") -def add_sub() -> Tuple[Dict[str, Any], int]: - try: - sub_info = get_sub_info_from_yt_url(request.form["url"]) - time_between_fetches = int(request.form["time_between_fetches"]) - except: - return {'error': 'Invalid data'}, 400 - sub = Subscription( - _id=sub_info["id"], - link=sub_info["link"], - title=sub_info["title"], - time_between_fetches=time_between_fetches, - ) - try: - sub.insert() - return sub_info_from_dict(sub.asdict()), 201 - except DuplicateKeyError: - return {'error': "Subscription %s already exists"%sub_info["id"] }, 409 - -@app.patch("/set-time-between-fetches/<id>") -def set_time_between_fetches(id: str) -> Tuple[Dict[str, Any], int]: - try: - time_between_fetches = int(request.form["time_between_fetches"]) - except: - return {'error': 'Invalid data'}, 400 - result = subscriptions.update_one( - {"_id": id}, - {"$set": {"time_between_fetches": time_between_fetches}} - ) - if result.modified_count: - return { - "_id": id, - "time_between_fetches": time_between_fetches, - }, 200 - return {'error': "Subscription %s not found"%id }, 404 - -@app.delete("/delete-sub/<id>") -def delete_sub(id: str) -> Tuple[Dict[str, Any], int]: - result = subscriptions.delete_one({"_id": id}) - if not result.deleted_count: - return {'error': "Subscription %s not found"%id }, 404 - return { "_id": id, }, 200 - -@app.patch("/set-viewed/<id>") -def set_viewed(id:str) -> Tuple[Dict[str, Any], int]: - viewed_time_str = request.form.get("viewed_time") - if viewed_time_str: - try: - viewed_time = datetime.fromisoformat(viewed_time_str) - except: - return {'error': 'Invalid data'}, 400 - else: - viewed_time = datetime.now(tz=UTC) - result = subscriptions.update_one( - {"_id": id}, - {"$set": {"last_viewed": viewed_time}} - ) - if result.modified_count: - return { - "_id": id, - "last_viewed": viewed_time, - }, 200 - return {'error': "Subscription %s not found"%id }, 404 diff --git a/wsgi/utils.py b/wsgi/utils.py deleted file mode 100644 index 2c9e1e5..0000000 --- a/wsgi/utils.py +++ /dev/null @@ -1,14 +0,0 @@ -from typing import Any, Dict, List -from components.subscriptions.main import Subscription -from components.subscriptions.typing import SubsDict -from components.videos import VideoTuple - -def vid_dicts_from_tuple_list(tuple_list: List[VideoTuple]) -> List[Dict[str, Any]]: - return [VideoTuple._make(vid)._asdict() for vid in tuple_list] - -def sub_info_from_dict(sub_dict: SubsDict) -> Dict[str, Any]: - return { - **sub_dict, - "videos": len(sub_dict["videos"]), - "new_vids": len(Subscription(**sub_dict).get_new_vids()), - } |
