ready more actions

Author: khasanjon-dev

main.py

@@ -1,33 +1,92 @@
-import re
+import base64
+import json
+import os
+import time
 
+import requests
+from dotenv import load_dotenv
+from requests import RequestException
 
-class TextProcessor:
-    def __init__(self, text):
-        self.text = text
+load_dotenv()
 
-    def extract_emails(self):
-        def extract_email(text):
-            pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.(ru|com|org|net)'
-            match = re.search(pattern, text)
-            if match:
-                return match.group(0)
-            return None
+ORG_USERNAME = os.environ.get("ORG_USERNAME")
+ORG_PASSWORD = os.environ.get("ORG_PASSWORD")
+ORG_GRANT_TYPE = os.environ.get("ORG_GRANT_TYPE")
+ORG_TOKEN_URL = os.environ.get("ORG_TOKEN_URL")
+ORG_CONSUMER_KEY = os.environ.get("ORG_CONSUMER_KEY")
+ORG_CONSUMER_SECRET = os.environ.get("ORG_CONSUMER_SECRET")
+ORG_COMPANY_URL = os.environ.get("ORG_COMPANY_URL")
 
-        emails = []
-        for t in self.text.split():
-            email = extract_email(t)
-            if email:
-                emails.append(email)
+cache = {}
 
-        emails = set(emails)
-        return sorted(emails)
 
+def get_token() -> dict | str:
+    """Get authentication token and cache it."""
+    if "org_token" in cache and cache["org_token"]["expires_at"] > time.time():
+        return cache["org_token"]["token"]
 
-input_text = input()
-processor = TextProcessor(input_text)
+    payload = {
+        "grant_type": ORG_GRANT_TYPE,
+        "username": ORG_USERNAME,
+        "password": ORG_PASSWORD,
+    }
+    auth_str = f"{ORG_CONSUMER_KEY}:{ORG_CONSUMER_SECRET}"
+    auth_base64 = base64.b64encode(auth_str.encode("utf-8")).decode("utf-8")
 
-found_emails = processor.extract_emails()
-if found_emails:
-    print("\n".join(found_emails))
-else:
-    print("Не найдено")
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Authorization": f"Basic {auth_base64}",
+    }
+    try:
+        response = requests.post(ORG_TOKEN_URL, headers=headers, data=payload)
+        response.raise_for_status()
+        token = response.json().get("access_token")
+        expires_in = response.json().get("expires_in", 3600)  # Standart 1 soat
+
+        if token:
+            cache["org_token"] = {
+                "token": token,
+                "expires_at": time.time() + expires_in  # Token muddati
+            }
+        return token
+    except RequestException as e:
+        return {"error": "Failed to obtain token", "details": str(e)}
+
+
+def get_company(inn: int | str) -> dict:
+    """Fetch company details by INN from an external service."""
+    payload = json.dumps({"tin": str(inn)})
+
+    token = get_token()
+    if isinstance(token, dict):
+        return token
+
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}",
+    }
+
+    try:
+        response = requests.post(ORG_COMPANY_URL, headers=headers, data=payload)
+        if response.status_code == 401:
+            token = get_token()
+            if isinstance(token, dict):
+                return token
+
+            headers["Authorization"] = f"Bearer {token}"
+            response = requests.post(ORG_COMPANY_URL, headers=headers, data=payload)
+
+        response.raise_for_status()
+        response_js = response.json()
+        return {
+            "name": response_js.get("name"),
+            "address": response_js.get("address"),
+            "oked": response_js.get("nc6Code"),
+            "ns_code": response_js.get("ns10Code"),
+        }
+
+    except RequestException as e:
+        return {"error": "Failed to fetch company details", "details": str(e)}
+
+
+print(get_company("200524845"))

requirements.txt

@@ -1 +1,3 @@
-black==25.1.0
+black==25.1.0
+requests==2.32.3
+python-dotenv==1.0.1