--- apiVersion: apps/v1 kind: Deployment metadata: name: worker-celery-beat namespace: flows labels: app: worker-celery-beat spec: progressDeadlineSeconds: 600 replicas: 1 revisionHistoryLimit: 10 selector: matchLabels: app: worker-celery-beat strategy: rollingUpdate: maxSurge: 25% maxUnavailable: 25% type: RollingUpdate template: metadata: labels: app: worker-celery-beat spec: affinity: podAntiAffinity: requiredDuringSchedulingIgnoredDuringExecution: - labelSelector: matchExpressions: - key: app operator: In values: - worker-celery-beat topologyKey: kubernetes.io/hostname containers: - command: - celery - -A - src.worker - beat - -l - DEBUG env: - name: WORKFLOWS_NOTIFICATIONS_FROM_EMAIL value: sarex@dsinv.ru - name: ENABLE_METRICS value: '0' - name: ENABLE_MAILGUN value: '0' - name: SMTP_HOST value: relay.dsinv.ru - name: SMTP_PORT value: '25' - name: FROM_EMAIL value: sarex@dsinv.ru - name: MAILGUN_HOST value: http:localhost:8000 - name: MAILGUN_API_KEY value: empty - name: NOTIFICATION_SETTINGS_USE_MAILGUN value: '0' - name: WORKFLOWS_HOST value: http://workflows-service.workflow.svc.cluster.local:8000/api/v1 - name: FLOWS_HOST value: http://backend-service.flows.svc.cluster.local:8000 - name: WORKFLOWS_NOTIFICATIONS_REGISTRY value: cr.yandex/crp3ccidau046kdj8g9q - name: WORKFLOWS_NOTIFICATIONS_SMTP_HOST value: relay.dsinv.ru - name: WORKFLOWS_NOTIFICATIONS_SMTP_PORT value: '25' - name: PLANNING_HOST value: http://backend-service.pm.svc.cluster.local:8000/api/pm/msp - name: PLANNING_USE value: 'True' - name: WORKFLOWS_NOTIFICATIONS_FROM_EMAI value: sarex@dsinv.ru - name: FLOWS_DB_HOST value: postgres-service.flows.svc.cluster.local - name: FLOWS_DB_DB valueFrom: secretKeyRef: key: database name: postgres-secret - name: FLOWS_DB_USERNAME valueFrom: secretKeyRef: key: username name: postgres-secret - name: FLOWS_DB_PASSWORD valueFrom: secretKeyRef: key: password name: postgres-secret - name: ISSUES_DB_HOST value: postgres-service.issues.svc.cluster.local - name: ISSUES_DB_DB valueFrom: secretKeyRef: key: database name: postgres-secret-issues - name: ISSUES_DB_USERNAME valueFrom: secretKeyRef: key: username name: postgres-secret-issues - name: ISSUES_DB_PASSWORD valueFrom: secretKeyRef: key: password name: postgres-secret-issues - name: DEBUG value: '0' - name: PG_PORT value: '5432' - name: FLOWS_DB_PORT value: '5432' - name: ISSUES_DB_PORT value: '5432' - name: DJANGO_HOST value: http://backend.django.svc.cluster.local:8000/api - name: DJANGO_BASE_HOST value: https://sarex.dsinv.ru - name: DJANGO_AUTH valueFrom: secretKeyRef: key: token name: django-secret - name: DOCUMENTATION_HOST value: http://documentations-service.documentations.svc.cluster.local:8080/internal/v1 - name: BASE_HOST value: https://sarex.dsinv.ru - name: RESOURCES_HOST value: http://resources-service.resources.svc.cluster.local:8000/ - name: TIMEOUT value: '120' - name: RESOURCE_URL value: http://resources-service.resources/api/v1 - name: GATEWAY_URL value: http://pdm-api.documentations.svc.cluster.local:8080/api/v1 - name: SYNC_RESOURCE_ID value: '1' - name: SERVICE_HOST value: https://sarex.dsinv.ru/flows/api/v1 - name: ENABLE_ANALYTICS value: '1' - name: ENABLE_CELERY value: '1' - name: CELERY_QUEUE value: flow - name: RABBITMQ_HOST value: rabbitmq-service.flows.svc - name: RABBITMQ_PORT value: '5672' - name: RABBITMQ_VHOST value: flow - name: PG_HOST value: postgres-service.flows.svc.cluster.local - name: PG_DB valueFrom: secretKeyRef: key: database name: postgres-secret - name: PG_LOGIN valueFrom: secretKeyRef: key: username name: postgres-secret - name: PG_PASSWORD valueFrom: secretKeyRef: key: password name: postgres-secret - name: DJANGO_TOKEN valueFrom: secretKeyRef: key: token name: django-secret - name: RABBITMQ_USERNAME valueFrom: secretKeyRef: key: username name: rabbitmq-secret - name: RABBITMQ_PASSWORD valueFrom: secretKeyRef: key: password name: rabbitmq-secret image: cr.yandex/crp3ccidau046kdj8g9q/flows-backend_worke:dsinv_700a79e5 imagePullPolicy: IfNotPresent name: worker-celery-beat ports: - containerPort: 8000 name: http protocol: TCP resources: requests: cpu: '1' memory: 512Mi terminationMessagePath: /dev/termination-log terminationMessagePolicy: File volumeMounts: - mountPath: /opt/src/worker/notifications_task.py name: notifications-task subPath: notifications_task.py - mountPath: /opt/src/worker/templates/daily_digest.html name: daily-digest subPath: daily_digest.html dnsPolicy: ClusterFirst imagePullSecrets: - name: dockerhub nodeSelector: name: generic restartPolicy: Always schedulerName: default-scheduler securityContext: {} terminationGracePeriodSeconds: 30 volumes: - configMap: defaultMode: 420 items: - key: notifications_task.py path: notifications_task.py name: notifications-task name: notifications-task - configMap: defaultMode: 420 items: - key: daily_digest.html path: daily_digest.html name: daily-digest name: daily-digest --- apiVersion: v1 data: daily_digest.html: "{% extends \"base_email.html\" %}\n\n{%- macro expiration_date(date, today) -%}\n{%- if date -%}\n{{\n date.strftime('%d.%m.%Y')\n}}\n{%- else -%}\nБез срока\n{%- endif -%}\n{%- endmacro %}\n\n{%- macro review_link(review, base_host) -%}\n{{ review['review_name'] | e }}\n{%- endmacro %}\n\n{%- block content %}\n

На сегодня в Sarex вашей реакции ждут следующие согласования:

\n\n{% if force_headers or len(expired_reviews) + len(ongoing_reviews) > 0 -%}\n

Согласования ({{ ongoing_reviews | length + expired_reviews | length }}):

\n{%- endif %}\n\n{% if force_headers or len(expired_reviews) > 0 -%}\n

Просроченные задачи ({{ expired_reviews | length }}):

\n
    \n {% for review in expired_reviews -%}\n

  1. \n Проект: {{ review['resource_name'] | e }}, Согласование: {{ review_link(review, base_host) }},
    \n Шаг: {{ review['step_name'] | e }}, Срок задачи: {{ expiration_date(review['expires_at'], today) }}, ID: (# {{ review['review_id'] }} )\n

    2. \n {%- endfor %}\n
\n{%- endif %}\n\n{% if force_headers or len(ongoing_reviews) > 0 -%}\n

Активные задачи ({{ ongoing_reviews | length }}):

\n
    \n {% for review in ongoing_reviews -%}\n

  1. \n Проект: {{ review['resource_name'] | e }}, Согласование: {{ review_link(review, base_host) }},
    \n Шаг: {{ review['step_name'] | e }}, Срок задачи: {{ expiration_date(review['expires_at'], today) }}, ID: (# {{ review['review_id'] }} )\n

    2. \n {%- endfor %}\n
\n{%- endif %}\n\n

С уважением,
команда Sarex

\n{% endblock %}\n" kind: ConfigMap metadata: name: daily-digest namespace: flows --- apiVersion: v1 data: notifications_task.py: "from collections import defaultdict\nfrom datetime import datetime\nfrom pathlib import Path\nfrom zoneinfo import ZoneInfo\n\nfrom jinja2 import Environment, FileSystemLoader\n\nfrom src.worker.celery import celery_app\nfrom src.worker.notifications_config import (\n DjangoClient,\n FlowsDatabase,\n ResourcesClient,\n IssuesDatabase,\n MailgunClient,\n NotificationGlobalSettings,\n WorkflowsNotificationsClient,\n)\n\n\n@celery_app.task(name=\"notify_users\")\ndef notify_users():\n django_client = DjangoClient()\n resources_client = ResourcesClient()\n flows_db = FlowsDatabase()\n flows_db_connection = flows_db.get_connection()\n issues_db = IssuesDatabase()\n issues_db_connection = issues_db.get_connection()\n\n users = django_client.get_users_with_enabled_notifications()\n users = {user[\"id\"]: user for user in users}\n if not users:\n flows_db_connection.close()\n issues_db_connection.close()\n return\n resources = resources_client.get_resources()\n resources = {resource[\"id\"]: resource[\"name\"] for resource in resources}\n\n with flows_db_connection.cursor() as cursor:\n cursor.execute(\n f\"\"\"\n SELECT\n r.id AS review_id,\n tq.reviewer_id AS user_id,\n r.resource_id AS resource_id,\n r.name AS review_name,\n s.name AS step_name,\n tq.end_date AS expires_at\n FROM task_queue tq\n JOIN review r\n ON r.id = tq.review_id\n JOIN step s\n ON s.id = r.current_step_id\n WHERE tq.reviewer_id IN ({\",\".join(str(x) for x in users.keys())})\n AND tq.is_active\n \"\"\"\n )\n result = cursor.fetchall()\n flows_db_connection.close()\n\n reviews_tasks = defaultdict(list)\n for review_task in result:\n review_task[\"resource_name\"] = resources.get(str(review_task[\"resource_id\"]), \"Без проекта\")\n reviews_tasks[review_task[\"user_id\"]].append(review_task)\n\n with issues_db_connection.cursor() as cursor:\n cursor.execute(\n f\"\"\"\n SELECT\n i.public_id AS issue_id,\n i.author_id AS user_id,\n i.resource_id AS resource_id,\n i.title AS name,\n i.completion_date AS expires_at\n FROM issues_issue i\n JOIN issues_issuestatus s\n ON s.id = i.status_id\n WHERE author_id IN ({\",\".join(str(x) for x in users.keys())})\n AND s.name NOT IN ('created', 'done')\n AND i.deleted_at IS NULL\n \"\"\"\n )\n authors_result = cursor.fetchall()\n cursor.execute(\n f\"\"\"\n SELECT\n i.public_id AS issue_id,\n ru.responsible_id AS user_id,\n i.resource_id AS resource_id,\n i.title AS name,\n i.completion_date AS expires_at\n FROM issues_issue i\n JOIN issues_responsibleuser ru\n ON ru.issue_id_id = i.public_id\n JOIN issues_issuestatus s\n ON s.id = i.status_id\n WHERE ru.responsible_id IN ({\",\".join(str(x) for x in users.keys())})\n AND s.name = 'created'\n AND i.deleted_at IS NULL\n \"\"\"\n )\n responsibles_result = cursor.fetchall()\n cursor.execute(\n \"\"\"\n SELECT\n i.public_id AS issue_id,\n i.responsible_entities AS responsible_entities,\n i.resource_id AS resource_id,\n i.title AS name,\n i.completion_date AS expires_at\n FROM issues_issue i\n JOIN issues_issuestatus s\n ON s.id = i.status_id\n WHERE i.responsible_entities IS NOT NULL\n AND s.name = 'created'\n AND\ \ i.deleted_at IS NULL\n \"\"\"\n )\n responsibles_sa_result = cursor.fetchall()\n issues_db_connection.close()\n\n issues_tasks = defaultdict(list)\n for issue_task in authors_result + responsibles_result:\n issue_task[\"resource_name\"] = resources.get(str(issue_task[\"resource_id\"]), \"Без проекта\")\n issues_tasks[issue_task[\"user_id\"]].append(issue_task)\n\n for issue in responsibles_sa_result:\n responsible_entities = issue[\"responsible_entities\"]\n positions = set()\n departments = set()\n for entity in responsible_entities[\"entities_list\"]:\n if entity[\"model_name\"] == \"position\":\n positions.add(str(entity[\"service_account_id\"]))\n elif entity[\"model_name\"] == \"department\":\n departments.add(str(entity[\"service_account_id\"]))\n if responsible_entities[\"operation\"] == \"AND\":\n if positions and departments:\n condition = lambda user_positions, user_departments: \\\n positions & user_positions and departments & user_departments\n elif positions:\n condition = lambda user_positions, _: \\\n positions & user_positions\n else:\n condition = lambda _, user_departments: \\\n departments & user_departments\n elif responsible_entities[\"operation\"] == \"OR\":\n condition = lambda user_positions, user_departments: \\\n positions & user_positions or departments & user_departments\n else:\n condition = lambda *_: False\n\n for user in users.values():\n user_positions, user_departments = set([p[\"service_account_id\"] for p in user[\"positions\"]]), \\\n set([d[\"service_account_id\"] for d in user[\"departments\"]])\n if condition(user_positions, user_departments):\n issue[\"resource_name\"] = resources.get(str(issue[\"resource_id\"]), \"Без проекта\")\n issues_tasks[user[\"id\"]].append(issue)\n\n jinja_env = Environment(loader=FileSystemLoader(Path(__file__).parent / \"templates\"))\n digest_template = jinja_env.get_template(\"daily_digest.html\")\n today = datetime.now(tz=ZoneInfo('UTC')).date()\n\n if NotificationGlobalSettings().use_mailgun:\n mailer = MailgunClient()\n else:\n mailer = WorkflowsNotificationsClient(job_name=\"Рассылка уведомлений: ежедневный дайджест\")\n\n for user_id, user in users.items():\n if not user[\"email\"]:\n continue\n reviews = reviews_tasks.get(user_id, [])\n issues = issues_tasks.get(user_id, [])\n if not (reviews or issues):\n continue\n\n expired_reviews = []\n ongoing_reviews = []\n for review in reviews:\n if bool(review[\"expires_at\"]) and review[\"expires_at\"].date() < today:\n expired_reviews.append(review)\n else:\n ongoing_reviews.append(review)\n\n # приклеиваем таймзону, чтобы все даты были offset-aware\n sorting_placeholder = datetime.max.replace(tzinfo=ZoneInfo('UTC'))\n expired_reviews.sort(key=lambda review: review[\"expires_at\"] or sorting_placeholder)\n ongoing_reviews.sort(key=lambda review: review[\"expires_at\"] or sorting_placeholder)\n issues.sort(key=lambda issue: issue[\"expires_at\"] or sorting_placeholder)\n\n html = digest_template.render(\n ongoing_reviews=ongoing_reviews,\n expired_reviews=expired_reviews,\n issues=issues,\n base_host=django_client.base_host,\n today=today,\n force_headers=True, # выводить ли хедеры, если сущностей ноль\n )\n\n mailer.send_email(\n send_to=user[\"email\"],\n subject=f\"Статус согласований в Sarex — {datetime.now(ZoneInfo('Europe/Moscow')):%d.%m.%Y}\",\n html=html,\n company_id=user[\"companies\"\ ][0],\n triggerer_user_id=user_id,\n )\n\n\n@celery_app.task(name=\"notify_admins_about_empty_steps\")\ndef notify_admins_about_empty_steps():\n django_client = DjangoClient()\n resources_client = ResourcesClient()\n flows_db = FlowsDatabase()\n flows_db_connection = flows_db.get_connection()\n \n if NotificationGlobalSettings().use_mailgun:\n mailer = MailgunClient()\n else:\n mailer = WorkflowsNotificationsClient(job_name=\"Рассылка уведомлений: отсутствуют согласующие\")\n\n\n admins = django_client.get_flow_admins()\n admins = {user[\"id\"]: user for user in admins}\n\n if not admins:\n flows_db_connection.close()\n return\n\n resources = resources_client.get_resources()\n resources = {resource[\"id\"]: resource[\"name\"] for resource in resources}\n\n resource2users = resources_client.get_users_by_resources()\n resource2admins: dict[str, list[dict]] = defaultdict(list)\n for resource, user_ids in resource2users.items():\n for user_id in user_ids:\n if user_id in admins:\n resource2admins[resource].append(admins[user_id])\n\n with flows_db_connection.cursor() as cursor:\n cursor.execute(\n \"\"\"\n SELECT\n r.id AS review_id,\n r.resource_id AS resource_id,\n r.name AS review_name,\n f.company_id AS company_id,\n s.name AS step_name\n FROM review r\n JOIN flow f\n ON f.id = r.flow_id\n JOIN step s\n ON s.id = r.current_step_id\n WHERE r.status = 'open'\n AND r.current_reviewers = '{}'\n AND f.is_active\n \"\"\"\n )\n result = cursor.fetchall()\n flows_db_connection.close()\n\n jinja_env = Environment(loader=FileSystemLoader(Path(__file__).parent / \"templates\"))\n email_template = jinja_env.get_template(\"daily_empty_step.html\")\n\n for review in result:\n render_values = {\n 'review_name': review['review_name'],\n 'resource_name': resources.get(str(review[\"resource_id\"]), \"Без проекта\"),\n 'review_id': review['review_id'],\n 'current_step': review['step_name'],\n 'link': f'{django_client.base_host}/reviews/{review[\"review_id\"]}'\n }\n\n body = email_template.render(render_values)\n to = [user['email'] for user in resource2admins[str(review['resource_id'])]]\n subject = (\n f\"Напоминание: Отсутствуют согласующие — {render_values['resource_name']}\"\n f\" — {render_values['review_id']}\"\n f\" — {render_values['current_step']}\"\n f\" — {render_values['review_name']}\"\n )\n\n if len(to) > 0:\n user_id = resource2admins[str(review['resource_id'])][0]['id']\n\n mailer.send_email(\n send_to=','.join(to),\n subject=subject,\n html=body,\n company_id=review['company_id'],\n triggerer_user_id=user_id,\n )\n" kind: ConfigMap metadata: name: notifications-task namespace: flows