# `tasks.py` generate from `tasks.ipynb` by `build.ipynb` from config import * from db import * from typing import Dict, Any, List, Optional, Callable, Union import email_task as email_task import vk as vk import tg as tg from utils import * from models import NotifyItem from auth import token_create import asyncio async def notify_in_profile(item: NotifyItem): return await notify_collection.update_one( { "user_id": item.user_id, "type": item.type_item, "title": item.title, "link": item.link, "text": item.text, }, { "$set": { "user_id": item.user_id, "type": item.type_item, "title": item.title, "text": item.text, "link": item.link, "read": True, "prod": PROD, } }, upsert=True, ) async def email_new_post(new_post: Dict): async for user in users_collection.find({'roles':'partner'}): login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/paintings/{new_post['_id']}')}' if PROD: email_task.notify.delay(user['email'], { 'text':'На сервисе опубликована новая картина', 'list':[[ {'label':'Название','text':new_post['title'], 'img':f"{FULL_DOMAIN}/b/posts/{new_post['medias_original'][0]}"}, {'label':'Звание художника','text':new_post['grade']}, {'label':'Город','text':new_post['city']}, {'label':'Размер','text':new_post['size']}, {'label':'Материал','text':new_post['material']}, {'label':'Жанр','text':new_post['genre']}, {'label':'Цена','text':new_post['price']}, ]], 'primary': {'href':login_link,'text':'Войти и посмотреть'}, 'title': f"от художника {new_post['author']}", 'login_link': login_link, }, 'Новая картина', 'main.html') async def update_social_post(post): pass async def publish_social_post(post): query = { 'active': True, "$or": [{'service': True}, {'user_id': post['user']['_id']}] } post_ref = {'_id': post['_id'], 'db': 'posts', 'col': 'posts'} medias = post.get("medias_original", []) platforms = [ ('tg', tg, lambda s: [ f"{FULL_PATH_DOMAIN}/a/{post['user']['_id']}" if should_sign_avatar(post, s) else None, *[f"{FULL_PATH_DOMAIN}/b/posts/{m}" for m in medias] ]), ('vk', vk, lambda s: [ post['user']['_id'] if should_sign_avatar(post, s) else None, *medias ]) ] for platform_type, client, get_attachments in platforms: if post.get(platform_type) != []: continue socials = [doc async for doc in social_collection.find({**query, 'type': platform_type})] socials = {item["id"]: item for item in socials}.values() for social in socials: text = parse_posts_text(post, platform_type, social.get('service', False)) sign_avatar = should_sign_avatar(post, social) attachments = [a for a in get_attachments(social) if a is not None] client.notify.delay(post_ref, text, attachments, bson_clear(social), sign_avatar) async def user_social_competition(competition): async for user in users_collection.find({'roles':'admin'}): # email_task.notify.delay if PROD: pass else: print(competition) class SocialContentHandler: """Base class for content type specific handlers""" def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: """Return reference for the content item""" raise NotImplementedError def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: """Return list of media file_ids""" raise NotImplementedError def get_text(self, item: Dict[str, Any]) -> str: """Parse text for the content""" raise NotImplementedError class CompetitionHandler(SocialContentHandler): def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: return {'_id': item['_id'], 'db': 'competitions', 'col': 'competitions'} def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: return [item.get("poster_file_id")] def get_text(self, item: Dict[str, Any]) -> str: return parse_text_competition(item) class PostHandler(SocialContentHandler): def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: return {'_id': item['_id'], 'db': 'posts', 'col': 'posts'} def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: # Adjust according to your Post schema return item.get("media_file_ids", []) or [item.get("main_image")] def get_text(self, item: Dict[str, Any]) -> str: return item.get("text", "") class NewsHandler(SocialContentHandler): def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: return {'_id': item['_id'], 'db': 'news', 'col': 'news'} def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: return item.get("media_ids", []) def get_text(self, item: Dict[str, Any]) -> str: return item.get("title", "") + "\n\n" + item.get("content", "") # Registry for handlers CONTENT_HANDLERS: Dict[str, SocialContentHandler] = { "competition": CompetitionHandler(), "post": PostHandler(), "news": NewsHandler(), # Add more: "user": UserProfileHandler(), etc. } class SocialPlatform: """Platform configuration""" def __init__( self, type_name: str, client: Any, get_attachments: Callable[[Dict], List[str]], sign_avatar_func: Optional[Callable] = None ): self.type = type_name self.client = client self.get_attachments = get_attachments self.sign_avatar_func = sign_avatar_func or should_sign_avatar # Platform registry - easy to extend PLATFORMS: Dict[str, SocialPlatform] = { "tg": SocialPlatform( type_name="tg", client=tg, get_attachments=lambda social, medias: [ f"{FULL_PATH_DOMAIN}/a/{social.get('user_id')}" if should_sign_avatar(social) else None, *[f"{FULL_PATH_DOMAIN}/b/competition/{m}" for m in medias if m] # generalize path ] ), "vk": SocialPlatform( type_name="vk", client=vk, get_attachments=lambda social, medias: [ social.get('user_id') if should_sign_avatar(social) else None, *medias ] ), # Add more platforms easily: # "ok": SocialPlatform(...), # "telegram_channel": ... } async def get_socials_for_content( content_type: str, item: Dict[str, Any], service_social: bool = False, user_social: bool = False, users_social: bool = False ) -> List[Dict]: """Get relevant social accounts based on rules""" query: Dict = {'active': True, "$or": []} if service_social: query["$or"].append({'service': True}) if user_social: query["$or"].append({'user_id': item.get('user', {}).get('_id')}) if users_social: query["$or"].extend([ {"competition": True}, {"competition": {"$exists": False}} ]) # Filter by platform types that exist in registry platform_types = list(PLATFORMS.keys()) cursor = social_collection.find({**query, 'type': {'$in': platform_types}}) socials = [doc async for doc in cursor] # Deduplicate by id return list({item["id"]: item for item in socials}.values()) async def send_to_social( content_type: str, item: Dict[str, Any], social: Dict[str, Any], platform: SocialPlatform ) -> bool: """Send one item to one social account with spam protection""" handler = CONTENT_HANDLERS.get(content_type) if not handler: print(f"Unknown content type: {content_type}") return False redis_key = f"nospamsends:{content_type}:{item['_id']}:{social['_id']}" already_sent = await redis_client.set( redis_key, "1", ex=8 * 60 * 60, # 8 hours nx=True ) if not already_sent: return False # already sent recently post_ref = handler.get_post_ref(item) medias = handler.get_medias(item) text = handler.get_text(item) sign_avatar = should_sign_avatar(item, social) if hasattr(handler, 'should_sign') else should_sign_avatar(item, social) attachments = [a for a in platform.get_attachments(social, medias) if a is not None] try: platform.client.notify.delay( post_ref, text, attachments, bson_clear(social), sign_avatar ) print(f"Queued {content_type} {item['_id']} to {platform.type} social {social['_id']}") return True except Exception as e: print(f"Failed to queue to {platform.type}: {e}") return False async def service_social_publish( content_type: str, item: Dict[str, Any], service_social: bool = False, user_social: bool = False, users_social: bool = False ): """ Generic social publishing service. Usage: await service_social_publish("competition", competition_dict) await service_social_publish("post", post_dict) """ if content_type not in CONTENT_HANDLERS: raise ValueError(f"Unsupported content type: {content_type}") socials = await get_socials_for_content( content_type, item, service_social, user_social, users_social ) tasks = [] for social in socials: platform_key = social.get("type") platform = PLATFORMS.get(platform_key) if not platform: continue tasks.append(send_to_social(content_type, item, social, platform)) if tasks: await asyncio.gather(*tasks, return_exceptions=True) # # Convenience wrappers # async def service_social_competition(competition: Dict, **kwargs): # return await service_social_publish("competition", competition, **kwargs) # async def service_social_post(post: Dict, **kwargs): # return await service_social_publish("post", post, **kwargs) # async def service_social_news(news: Dict, **kwargs): # return await service_social_publish("news", news, **kwargs) async def service_social_exhibition(exhibition): query = { 'active': True, "$or": []} query["$or"].append({'service': True}) query["$or"].append({'user_id': exhibition['user']['_id']}) post_ref = {'_id': str(exhibition['_id']), 'db': 'exhibition', 'col': 'exhibition'} medias = [exhibition.get("poster_orig_id")] platforms = [ ('tg', tg, lambda s: [ # f"{FULL_PATH_DOMAIN}/a/{exhibition['user']['_id']}" if should_sign_avatar(exhibition, s) else None, *[f"{FULL_PATH_DOMAIN}/b/exhibition/{m}" for m in medias] ]), ('vk', vk, lambda s: [ # exhibition['user']['_id'] if should_sign_avatar(exhibition, s) else None, *medias ]) ] for platform_type, client, get_attachments in platforms: if exhibition.get(platform_type) != []: continue socials = [doc async for doc in social_collection.find({**query, 'type': platform_type})] socials = {item["id"]: item for item in socials}.values() for social in socials: text = parse_text_exhibition(exhibition) sign_avatar = should_sign_avatar(exhibition, social) attachments = [a for a in get_attachments(social) if a is not None] redis_key = f"nospamsends:{platform_type}:{exhibition['_id']}:{social['_id']}" already_sent = await redis_client.set( redis_key, 1, ex=8*60*60, nx=True ) print(f"nospamsends:{platform_type}:{exhibition['_id']}:{social['_id']}") if already_sent: print(social) client.notify.delay(post_ref, text, attachments, bson_clear(social), False) async def service_social_competition(competition, serviceSocial, userSocial, usersSocial): query = { 'active': True, "$or": []} if serviceSocial: query["$or"].append({'service': True}) if userSocial: query["$or"].append({'user_id': competition['user']['_id']}) if usersSocial: query["$or"].append({"competition": True}) query["$or"].append({"competition": {"$exists": False}}) post_ref = {'_id': str(competition['_id']), 'db': 'competitions', 'col': 'competitions'} medias = [competition.get("poster_file_id")] platforms = [ ('tg', tg, lambda s: [ # f"{FULL_PATH_DOMAIN}/a/{competition['user']['_id']}" if should_sign_avatar(competition, s) else None, *[f"{FULL_PATH_DOMAIN}/b/competition/{m}" for m in medias] ]), ('vk', vk, lambda s: [ # competition['user']['_id'] if should_sign_avatar(competition, s) else None, *medias ]) ] for platform_type, client, get_attachments in platforms: socials = [doc async for doc in social_collection.find({**query, 'type': platform_type})] socials = {item["id"]: item for item in socials}.values() for social in socials: text = parse_text_competition(competition) sign_avatar = should_sign_avatar(competition, social) attachments = [a for a in get_attachments(social) if a is not None] redis_key = f"nospamsends:{platform_type}:{competition['_id']}:{social['_id']}" already_sent = await redis_client.set( redis_key, 1, ex=8*60*60, nx=True ) print(f"nospamsends:{platform_type}:{competition['_id']}:{social['_id']}") if already_sent: print(social) client.notify.delay(post_ref, text, attachments, bson_clear(social), False) async def email_new_user(new_user): async for user in users_collection.find({'roles':'admin'}): login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/artists/{new_user['_id']}')}' role = {'user':'Художник','galerist':'Галерея',"reporter":'Репортер'}[new_user['roles'][0]] if PROD: email_task.notify.delay(user['email'], { 'list':[[ {'label':'ФИО','text': new_user['female_name_lastname'], 'img':f"{FULL_DOMAIN}/a/{new_user['_id']}"}, {'label':'Звание','text': new_user['grade']} if new_user['roles'][0] != 'galerist' else {}, {'label':'Город проживания','text': new_user['living_place']} ]], 'text': f'', 'primary': {'href':login_link,'text':'Войти и посмотреть'}, 'title': f"{role} {new_user['username']}", 'login_link': login_link }, 'Новый пользователь', 'main.html') async def invite_users_to_competitions(doc): new_notify = [] query = { "roles": {"$in": ["user"]}, "$or": [ {"email_subscribed": True}, {"email_subscribed": {"$exists": False}} ] } async for user in users_collection.find(query): if not user.get("email"): continue redis_key = f"nospam:{user['email']}:{doc['_id']}" already_sent = await redis_client.set( redis_key, 1, ex=24*60*60, nx=True ) if already_sent: new_notify.append( notify_in_profile(NotifyItem( user_id=str(user['_id']), type_item='competition', title='Приглашение на конкурс', link=f'/profile/competition/{doc.get("id") or doc.get("_id")}', text=f'«{doc['title']}» Организатор {doc['user']['username']}' )) ) if PROD: email_task.notify.delay(user['email'], {'competition': bson_clear(doc), 'login_link': f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/competition/{doc.get("id") or doc.get("_id")}')}'}, 'Конкурс', 'competition.html') await asyncio.gather(*new_notify) async def accept_competition_req(competition, user_id): user = await users_collection.find_one({'_id': ObjectId(user_id)},{'email':1}) login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/profile/invitation')}' if PROD: email_task.notify.delay(user['email'], { 'text':f'на конкурсе {competition['title']}', 'list':[[ ]], 'primary': {'href':login_link,'text':'Войти и посмотреть'}, 'title': f"Вы прошли отбор", 'login_link': login_link, }, 'Приглашение', 'main.html') await notify_in_profile(NotifyItem(user_id=user_id, type_item='invation', title='Приглашение', link='/profile/invitation', text=f'Вы прошли отбор на конкурсе {competition['title']}')) async def get_sender(post_id): delivery_cursor = delivery_collection.find( { 'post_id': post_id,"prod": PROD, 'status': { '$nin': ["Ожидает отправки", "Ожидает принятия условий", "Третье"] } }, {'receiver': 1} ).sort('_id', 1) # 1 = ascending, -1 = descending (newest first) delivery = await delivery_cursor.limit(1).to_list(length=1) delivery = delivery[0] if delivery else {} return delivery.get('receiver') from typing import List from bson import ObjectId async def accept_invitations(invitations_id: str, posts_ids: List[str]): invitation = await invitations_collection.find_one_and_update( {"_id": ObjectId(invitations_id)}, {"$set": {"accept": True}}, return_document=True, ) invitation_posts = invitation.get('posts', []) or [] invitation_post_ids = [str(doc['_id']) for doc in invitation_posts] selected_set = set(str(x) for x in posts_ids) selected_ids = [pid for pid in invitation_post_ids if pid in selected_set] other_ids = [pid for pid in invitation_post_ids if pid not in selected_set] if selected_ids: await delivery_collection.update_many( {'post_id': {'$in': selected_ids}}, {'$set': {'status': 'Ожидает отправки'}} ) if other_ids: await delivery_collection.update_many( {'post_id': {'$in': other_ids}}, {'$set': {'status': 'Приглашение откланено'}} # как вы написали ) await notify_in_profile(NotifyItem( user_id=invitation.get('exhibition',{}).get("user",{}).get('_id'), type_item='c', title=f'Выставка "{invitation.get('exhibition',{}).get("title")}"', link=f'/profile/delivery', text=f'Художник принял условия' )) async def exhibition_send_invitations(exhibition_id): exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)}) exhibition_owner = exhibition.get('user') obj_ids = [ObjectId(x) for x in exhibition.get('posts', [])] posts = await posts_collection.find({"_id": {"$in": obj_ids}}).to_list() if obj_ids else [] new_posts = {} all_posts = {} for post in posts: post_id = str(post.get('_id')) user = display_user(post.get('user')) doc = await delivery_collection.update_one( { "exhibition_id": str(exhibition.get('_id')),"post_id": post_id, "status": "Ожидает принятия условий", "type": "exhibition", },{ "$set":{ "sender": user, "receiver": display_user(exhibition_owner), "exhibition_id": str(exhibition.get('_id')), "address": exhibition.get('address'), "post_id": post_id, "prod": PROD, "img": post.get('medias',[0])[0], "title": post.get('title'), "status": "Ожидает принятия условий", "type": "exhibition", "updated_at": datetime.utcnow() } }, upsert=True) if doc.upserted_id: if (user_id:=user.get('_id')) in new_posts: new_posts[user_id].append(post) else: new_posts.update({user_id: [post]}) if (user_id:=user.get('_id')) in all_posts: all_posts[user_id].append(post) else: all_posts.update({user_id: [post]}) print(all_posts) if all_posts: for user_id, posts in all_posts.items(): doc = await invitations_collection.update_one( {'exhibition_id': str(exhibition.get('_id')),'user_id': user_id,'type': 'Выставка'}, {"$set":{'exhibition': bson_clear(exhibition), 'exhibition_id': str(exhibition.get('_id')), 'accept': False, 'user':display_user(exhibition_owner), 'user_id': user_id, 'posts': posts, 'prod': PROD, 'type': 'Выставка', 'item': exhibition.get('title')}}, upsert=True ) if new_posts: for user_id, posts in new_posts.items(): await notify_in_profile(NotifyItem( user_id=user_id, type_item='invitation', title='Приглашение', link=f'/profile/invitation', text=f'Приглашаем на выставку "{exhibition.get("title")}"' )) user = posts[0].get('user') token = await token_create( str(user['_id']), url=f'/profile/invitation#main' ) login_link = f'{FULL_DOMAIN}/login/{token}' email_data = { 'text': 'Приглашаем на выставку со следующими картинами: ', 'list': [ [ {'label': 'Название', 'text': item.get('title', ''), 'img':f"{FULL_DOMAIN}/b/posts/{item['medias_original'][0]}"}, ] for item in posts ], 'primary': { 'href': login_link, 'text': 'Войти и посмотреть' }, 'title': exhibition.get('title', 'Выставка'), 'login_link': login_link, } if PROD: email_task.notify.delay( user['email'], email_data, f'Приглашение на выставку {exhibition.get('title')}', 'main.html' ) async def update_competition_req(competition, user, posts): competition_id = competition.get('_id') if not user or not posts or not competition_id: return posts_sorted = sorted(posts, key=lambda p: p.get('status') != 'Проходит') # Генерация login link token = await token_create( str(user['_id']), url=f'/profile/delivery#main' ) login_link = f'{FULL_DOMAIN}/login/{token}' if True: competition_owner = competition.get('user', {}) # === Создание доставок === post_ids_to_mark = [] # для последующего обновления delivery: True post_ids_to_remove_delivery = [] for post in posts: post_id = str(post.get('_id')) if post.get('status') == 'Проходит': sender = await get_sender(post_id) or display_user(user) delivery_collection.update_one({ "competition_id": competition_id,"post_id": post_id, "status": "Ожидает отправки", "sender._id": sender.get('_id'), "receiver._id": competition_owner.get('_id') },{ "$set":{ "sender": sender, "user": display_user(user), "receiver": display_user(competition_owner), "competition_id": competition_id, "address": competition.get('address'), "post_id": post_id, "prod": PROD, "img": post.get('medias',[0])[0], "title": post.get('title'), "status": "Ожидает отправки", "type": "competition_reward", "updated_at": datetime.utcnow() } }, upsert=True) post_ids_to_mark.append(post_id) elif post.get('status') != 'Проходит': post_ids_to_remove_delivery.append(post_id) if post_ids_to_remove_delivery: await delivery_collection.delete_many({ "competition_id": competition_id, "status": "Ожидает отправки", "post_id": {"$in": post_ids_to_remove_delivery} }) await notify_in_profile(NotifyItem( user_id=str(user['_id']), type_item='invitation', title='Приглашение', link=f'/profile/delivery', text=f'Вы прошли отбор на конкурсе "{competition.get("title")}"' )) all_allowed = all( post.get('status') in ('Проходит', 'На рассмотрении') for post in posts ) has_approved = any(p.get('status') == 'Проходит' for p in posts) # Email-уведомление if user.get('email') and all_allowed and has_approved: print([i['title'] for i in posts_sorted]) if PROD and user.get('email') and all_allowed and has_approved: email_data = { 'text': 'Статус заявки обновлен', 'list': [ [ {'label': 'Название', 'text': item.get('title', ''), 'img':f"{FULL_DOMAIN}/b/posts/{item['medias_original'][0]}"}, {'label': 'Статус', 'text': item.get('status', '')} ] for item in posts_sorted ], 'primary': { 'href': login_link, 'text': 'Войти и посмотреть' }, 'title': competition.get('title', 'Конкурс'), 'login_link': login_link, } email_task.notify.delay( user['email'], email_data, 'Обновление статуса заявки на конкурс', 'main.html' ) async def new_competiton_req(request): competition = await competitions_col.find_one({'_id':ObjectId(request['competition_id'])}) user = competition['user'] if competition.get('type_') == 'artist': link = f'/profile/partner/competition/{request['competition_id']}/{request['user']['_id']}' else: link = f'/profile/partner/competition/{request['competition_id']}' login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=link)}' print(user) await notify_in_profile(NotifyItem( user_id=user['_id'], type_item='competition_req', title='Новая заявка на конкурс', link=link, text=f'«{competition['title']}» От {request['user']['username']}' )) if PROD: email_task.notify.delay(user['email'], { 'list':[[{'label':'Название','text':i['title'], 'img':f"{FULL_DOMAIN}/b/posts/{i['medias'][0]}"},{'label':'Размер','text':i['size']},{'label':'Описание','text':i['description']}] for i in request['posts']], 'text': f'Картины на конкурс «{competition['title']}»:', 'primary': {'href':login_link,'text':'Войти и посмотреть'}, 'title': f"От {request['user']['username']}", 'login_link': login_link }, 'Новая заявка на конкурс', 'main.html')