fix force push

watchdoc/Pipfile

@@ -12,6 +12,7 @@ google-auth-oauthlib = "*"
 jinja2 = "*"
 uvicorn = {extras = ["standard"], version = "*"}
 pydantic = "*"
+yadisk = "*"
 
 [dev-packages]
 

watchdoc/src/config.py

@@ -35,3 +35,7 @@
 EMAIL_HOST_USER = os.environ["EMAIL_HOST_USER"]
 EMAIL_HOST_PASSWORD = os.environ["EMAIL_HOST_PASSWORD"]
 DEFAULT_FROM_EMAIL = f"Даль ВУЦ ВШЭ <{EMAIL_HOST_USER}>"
+
+# ------------------------------------------------------------------------------
+# YaDisk
+YADISK_TOKEN = os.environ["YADISK_TOKEN"]

watchdoc/src/disk.py

@@ -0,0 +1,30 @@
+import yadisk
+
+
+class DiskService:
+    """Yandex Disk API wrapper.
+
+    Usage:
+        ds = DiskService()
+    """
+
+    def __init__(self, token) -> None:
+        self._service = yadisk.YaDisk(token=token)
+
+    def obtain_folder(self, path, **kwargs):
+        if not self._service.exists(path):
+            self._service.mkdir(path, **kwargs)
+        return path
+
+    def upload_docx(self, local_path, remote_path, **kwargs):
+        with open(local_path, "rb") as f:
+            return self._service.upload(f, remote_path, **kwargs)
+
+    def delete_file(self, path, **kwargs):
+        self._service.remove(path, **kwargs)
+
+    def read_share_folder_with_anyone(self, remote_path):
+        self._service.publish(remote_path)
+        public_key = self._service.get_meta(remote_path)["public_key"]
+        public_resource = self._service.get_public_meta(public_key)
+        return public_resource["public_url"]

watchdoc/src/proto.py

@@ -75,4 +75,3 @@ class Applicant(Person):
     recruitment_office: str
 
     milspecialty: Milspecialty
-

watchdoc/src/service.py

@@ -1,13 +1,12 @@
 import base64
+from pathlib import PurePath
 import logging as log
 
 import jinja2
 
 from docxtpl import DocxTemplate
 
-from auth import obtain_credentials
 from email_service import EmailService
-from drive import DriveService
 from campuses import Campus
 from family import RelativeType
 from proto import Applicant
@@ -19,6 +18,7 @@
     EMAIL_HOST_USER,
     EMAIL_HOST_PASSWORD,
     EMAIL_USE_TLS,
+    YADISK_TOKEN,
 )
 
 from email_utils import create_message
@@ -27,10 +27,11 @@
     month_to_russian_title,
     today,
 )
+from src.disk import DiskService
 
-WATCHDOC_FOLDER: str = "watchdoc"
-APPLICANTS_FOLDER: str = "Абитуриенты"
-CAMPUSES_FOLDER: str = "Кампусы"
+WATCHDOC_FOLDER: PurePath = PurePath("watchdoc")
+APPLICANTS_FOLDER: PurePath = WATCHDOC_FOLDER / "Абитуриенты"
+CAMPUSES_FOLDER: PurePath = APPLICANTS_FOLDER / "Кампусы"
 
 DUMMY_IMAGE = TEMPLATES_DIR / "dummy.png"
 
@@ -52,16 +53,19 @@ class WatchDocService:
         folder_link = wds.upload_documents(applicant)
         wds.notify(applicant, folder_link)
     """
+
     def generate_documents(self, applicant: Applicant):
         applicant_dir = GENERATED_DIR / applicant.contact_info.corporate_email
         applicant_dir.mkdir(exist_ok=True)
 
         parents = [
-            r for r in applicant.family
+            r
+            for r in applicant.family
             if r.type in [RelativeType.FATHER, RelativeType.MOTHER]
         ]
         siblings = [
-            r for r in applicant.family
+            r
+            for r in applicant.family
             if r.type in [RelativeType.BROTHER, RelativeType.SISTER]
         ]
 
@@ -92,25 +96,18 @@ def upload_documents(self, applicant: Applicant):
         folder_name = f"{applicant.full_name} {email}"
 
         applicant_folder = self.ds.obtain_folder(
-            name=folder_name,
-            parents=[self._campus_folders[campus]["id"]],
+            self._campus_folders[campus] / folder_name
         )
 
         for (_, rus) in DOCUMENTS:
-            body = {
-                "name": rus,
-                "parents": [applicant_folder["id"]],
-            }
             self.ds.upload_docx(
-                body=body,
-                path=GENERATED_DIR / email / rus,
+                local_path=GENERATED_DIR / email / rus,
+                remote_path=applicant_folder / rus,
+                overwrite=True,
             )
 
-        self.ds.read_share_folder_with_anyone(
-            folder_id=applicant_folder["id"],
-        )
-
-        return applicant_folder["webViewLink"]
+        link = self.ds.read_share_folder_with_anyone(applicant_folder)
+        return link
 
     def notify(self, applicant: Applicant, folder_link: str):
         email = applicant.contact_info.corporate_email
@@ -124,20 +121,13 @@ def __init__(self) -> None:
         # Templates
         self.jinja_env = self._init_jinja_env()
 
-        # Credentials
-        credentials = obtain_credentials()
-
         # Email
         self.email_service = EmailService(
-            EMAIL_HOST,
-            EMAIL_PORT,
-            EMAIL_HOST_USER,
-            EMAIL_HOST_PASSWORD,
-            EMAIL_USE_TLS
+            EMAIL_HOST, EMAIL_PORT, EMAIL_HOST_USER, EMAIL_HOST_PASSWORD, EMAIL_USE_TLS
         )
 
-        # Drive
-        self.ds = DriveService(credentials)
+        # Disk
+        self.ds = DiskService(token=YADISK_TOKEN)
         self._watchdoc_folder = self._init_watchdoc_folder()
         self._applicants_folder = self._init_applicants_folder()
         self._campuses_folder = self._init_campuses_folder()
@@ -150,34 +140,22 @@ def _init_jinja_env(self):
         return env
 
     def _init_watchdoc_folder(self):
-        folder = self.ds.obtain_folder(name=WATCHDOC_FOLDER)
-        log.info(f"`{WATCHDOC_FOLDER}` web view link: "
-                 f"{folder['webViewLink']}")
+        folder = self.ds.obtain_folder(WATCHDOC_FOLDER)
+        log.info(f"Created applicant folder in `{folder}`")
         return folder
 
     def _init_applicants_folder(self):
-        folder = self.ds.obtain_folder(
-            name=APPLICANTS_FOLDER,
-            parents=[self._watchdoc_folder["id"]],
-        )
-        log.info(f"`{APPLICANTS_FOLDER}` web view link: "
-                 f"{folder['webViewLink']}")
+        folder = self.ds.obtain_folder(APPLICANTS_FOLDER)
+        log.info(f"Created applicant folder in `{folder}`")
         return folder
 
     def _init_campuses_folder(self):
-        folder = self.ds.obtain_folder(
-            name=CAMPUSES_FOLDER,
-            parents=[self._applicants_folder["id"]],
-        )
-        log.info(f"`{CAMPUSES_FOLDER}` web view link: "
-                 f"{folder['webViewLink']}")
+        folder = self.ds.obtain_folder(CAMPUSES_FOLDER)
+        log.info(f"Created applicant folder in `{folder}`")
         return folder
 
     def _init_campus_folders(self):
         folders = {}
         for abbr, campus in Campus.choices():
-            folders[abbr] = self.ds.obtain_folder(
-                name=campus,
-                parents=[self._campuses_folder["id"]],
-            )
+            folders[abbr] = self.ds.obtain_folder(CAMPUSES_FOLDER / campus)
         return folders