недоделал рассылку по подходящим

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+/.venv

.vscode/launch.json

@@ -0,0 +1,24 @@
+{
+  // Use IntelliSense to learn about possible attributes.
+  // Hover to view descriptions of existing attributes.
+  // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "Authorize",
+      "type": "python",
+      "request": "launch",
+      "module": "hh_applicant_tool",
+      "args": ["-vv", "authorize"],
+      "justMyCode": true
+    },
+    {
+      "name": "Apply Jobs",
+      "type": "python",
+      "request": "launch",
+      "module": "hh_applicant_tool",
+      "args": ["-vv", "apply-jobs"],
+      "justMyCode": true
+    }
+  ]
+}

README.md

@@ -0,0 +1,134 @@
+# HH Applicant Tool
+
+Утилита для автоматизации таких действий на HH как автоматический отклик на подходящие вакансии.
+
+Системные требования:
+
+- socat
+- python >= 3.10
+
+Нужную версию можно поставить через asdf/pyenv, а вот socat придется доставить.
+
+Данная утилита не может работать от root. Так что любители кало-линукса идут нахуй. Туда же идут пользователи Windows, так как я им не пользуюсь, но никто вам не мешает его добавить.
+
+Предыстория.
+
+В общем у меня был один знакомый знакомого, который работал ХЕРом. Этот чувак не заморачивался с чтением резюме, а тупо скриптами рассылал предложения о работе... Бывают, конечно, филологини, которые не могут отлчитьб Java от JavaScript, но я думаю, что в значительном числе ситуаций тдет именно о рассылках... И я просто перенял эту тактику. Долгое время я делал массовые заявки с помощью консоли браузера:
+
+```js
+$$('[data-qa="vacancy-serp__vacancy_response"]').forEach((el) => el.click());
+```
+
+И оно работает, хоть и не идеально. Я даже пробовал автоматизировать рассылки через `p[yu]ppeeter`, пока не прочитал [документацию](https://github.com/hhru/api). И не обнаружил много там интересных методов, использование которых через кукловода проблематично...
+
+Данное приложение работает через CLIENT_ID и CLIENT_SECRET, полученные мною путем [декомпиляции официального приложения для Android](https://gist.github.com/s3rgeym/eee96bbf91b04f7eb46b7449f8884a00). Не знаю считается это взломом или нет.
+
+Установка:
+
+```bash
+# Через pypi
+# Можно использовать и обычный pip
+$ pipx install hh-applicant-tool
+
+# Если хочется использовать самую последнюю версию, то можно установить ее через git
+$ pipx install git+https://github.com/s3rgeym/hh-applicant-tool
+```
+
+Помощь:
+
+```bash
+$ hh-applicant-tool
+...
+
+$ hh-applicant-tool apply-jobs -h
+usage: hh-applicant-tool apply-jobs [-h] [--resume-id RESUME_ID] [--message-list MESSAGE_LIST]
+
+Откликнуться на все подходящие вакансии
+
+options:
+  -h, --help            show this help message and exit
+  --resume-id RESUME_ID
+                        Идентефикатор резюме
+  --message-list MESSAGE_LIST
+                        Путь до файла, где хранятся сообщения для отклика на вакансии. Каждое сообщение — с новой строки. В сообщения можно использовать плейсхолдеры типа
+                        %(name)s
+```
+
+Для начала нужно добавить обработчик протокола `hhandroid`, который используется Android-приложением для усложнения жизни честным автоматизаторам:
+
+```bash
+$ hh-applicant-tool -vv add-handler
+[I] saved /home/sergey/.local/share/applications/hhandroid.desktop
+✅ Обработчик добавлен!
+```
+
+Флаг `-v` используется для вывода отладочной информации. Два таких флага выводят всю возможную.
+
+Авторизуемся:
+
+```bash
+$ hh_applicant_tool -vv authorize
+Пробуем открыть в браузере: https://hh.ru/oauth/authorize?client_id=HIOMIAS39CA9DICTA7JIO64LQKQJF5AGIK74G9ITJKLNEDAOH5FHS5G1JI7FOEGD&response_type=code
+Авторизуйтесь и нажмите <<Подтвердить>>
+[I] 🚀 Стартуем TCP-сервер по адресу unix:///tmp/hhandroid.sock
+Gtk-Message: 20:52:59.280: Failed to load module "canberra-gtk-module"
+Gtk-Message: 20:52:59.975: Failed to load module "canberra-gtk-module"
+[54:54:0305/205300.038812:ERROR:gl_factory.cc(128)] Requested GL implementation (gl=desktop-gl,angle=none) not found in allowed implementations: [(gl=egl-angle,angle=default),(gl=egl-gles2,angle=none),(gl=egl-angle,angle=swiftshader)].
+[54:54:0305/205300.041723:ERROR:viz_main_impl.cc(186)] Exiting GPU process due to errors during initialization
+Opening in existing browser session.
+[D] hhandroid://oauthresponse?code=99Q9G1RII75D8R2FTU06BF2FDNI7JF16MGBIB4OEQ973819OOJI90S69I1CL9U96
+[D] POST https://hh.ru/oauth/token 200
+[D] Сохраняем токен
+🔓 Авторизация прошла успешно!
+```
+
+![image](https://user-images.githubusercontent.com/12753171/222978533-ed30a918-ed15-4a81-a8c2-f083e8469c16.png)
+
+Тут надо выбирать `Open xdg-open`.
+
+После смотрим в консоль и закрываем вкладку. Она сама не закроется.
+
+При авторизации можно указать `redirect_uri`, но любые адреса кроме того, что с протоколом `hhandroid`, будут приводить к ошибке:
+
+![](https://user-images.githubusercontent.com/12753171/222870516-b29f2417-d11a-4122-8291-7d440a422a31.png)
+
+Поэтому и нужно добавление обработчика кастомного протокола. На том шаге создается `desktop` файл, где в секции `Exec` всего пару команд для того чтобы записать полученный uri в сокет. TCP-сервер, который запускается при авторизации, как раз слушает этот сокет... Идея была хорошей для защиты от скрипт-кидис, но Сему ведь этим не остановить (c).
+
+Проверка:
+
+```bash
+$ hh-applicant-tool whoami
+{
+  "auth_type": "applicant",
+  "counters": {
+    "new_resume_views": 1488,
+    "resumes_count": 1,
+    "unread_negotiations": 228
+  },
+  "email": "vasya.pupkin@gmail.com",
+  "employer": null,
+  "first_name": "Вася",
+  "id": "1234567890",
+  "is_admin": false,
+  "is_anonymous": false,
+  "is_applicant": true,
+  "is_application": false,
+  "is_employer": false,
+  "is_in_search": true,
+  "last_name": "Пупкин",
+  "manager": null,
+  "mid_name": null,
+  "middle_name": null,
+  "negotiations_url": "https://api.hh.ru/negotiations",
+  "personal_manager": null,
+  "phone": "79012345678",
+  "profile_videos": {
+    "items": []
+  },
+  "resumes_url": "https://api.hh.ru/resumes/mine"
+}
+```
+
+Утилита использует систему плагинов. Все они лежат в `hh_applicant_tool/operations`. Модули расположенные там автоматически добавляются как доступные операции. За основу для своего плагина можно взять `whoami.py`.
+
+Отдельные замечания у меня к API. Оно какое-то кривое. Какой долбоеб придумал при создании объекта отдавать пустой ответ (по REST должен быть созданный объект) либо вообще перенаправлять на полную версию сайта? Так же в ответах сервера нет Content-Length. Я так понял там какой-то прокси оборачивает все запросы и отдает всегда TE Chunked. А еще он возвращает 502 ошибку, когда бекенд на Java падает (я почти уверен, что HH написан на ней). А [язык запросов](https://hh.ru/article/1175) мне понравился. Можно что-то типа этого использовать `NOT (!ID:123 OR !ID:456 OR !ID:789)` что бы отсеить какие-то вакансии.

hh_applicant_tool/__main__.py

@@ -0,0 +1,6 @@
+import sys
+
+from .main import main
+
+if __name__ == "__main__":
+    sys.exit(main())

hh_applicant_tool/api.py

@@ -0,0 +1,219 @@
+"""https://api.hh.ru/openapi/redoc"""
+from __future__ import annotations
+
+import dataclasses
+import json
+import logging
+import time
+from copy import deepcopy
+from dataclasses import dataclass
+from functools import partialmethod
+from threading import Lock
+from typing import Any, Literal, TypedDict
+from urllib.parse import urlencode
+
+import requests
+from requests import Response, Session
+
+from .contsants import HHANDROID_CLIENT_ID, HHANDROID_CLIENT_SECRET
+
+logger = logging.getLogger(__package__)
+
+
+class BaseException(Exception):
+    def __init__(self, data: dict[str, Any]) -> None:
+        self._raw = deepcopy(data)
+
+    def __getattr__(self, name: str) -> Any:
+        try:
+            return self._raw[name]
+        except KeyError as ex:
+            raise AttributeError(name) from ex
+
+    def __str__(self) -> str:
+        return str(self._raw)
+
+
+class BadRequest(BaseException):
+    @property
+    def limit_exceeded(self) -> bool:
+        return any(x["value"] == "limit_exceeded" for x in self.errors)
+
+
+class Forbidden(BaseException):
+    pass
+
+
+class ResourceNotFound(BaseException):
+    pass
+
+
+# По всей видимости, прокси возвращает, когда их бекенд на Java падает
+# {'description': 'Bad Gateway', 'errors': [{'type': 'bad_gateway'}], 'request_id': '<md5 хеш>'}
+class BadGateaway(BaseException):
+    pass
+
+
+# Thread-safe
+@dataclass
+class BaseClient:
+    _: dataclasses.KW_ONLY
+    base_url: str
+    request_body_json: bool = False
+    # TODO: сделать генерацию User-Agent'а как в приложении
+    user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
+    session: Session | None = None
+    previous_request_time: float = 0.0
+
+    def __post_init__(self) -> None:
+        self.lock = Lock()
+        if not self.session:
+            self.session = session = requests.session()
+            session.headers.update(
+                {
+                    **self.additional_headers(),
+                    "User-Agent": self.user_agent,
+                }
+            )
+
+    def additional_headers(self) -> dict[str, str]:
+        return {}
+
+    def request(
+        self,
+        method: Literal["GET", "POST", "PUT", "DELETE"],
+        endpoint: str,
+        params: dict | None = None,
+        delay: float = 0.3,
+        **kwargs: Any,
+    ) -> dict:
+        assert method == method.upper()
+        params = dict(params or {})
+        params.update(kwargs)
+        url = self.resolve_url(endpoint)
+        with self.lock:
+            # На серваке какая-то анти-DDOS система
+            if (delay := delay - time.monotonic() + self.previous_request_time) > 0:
+                logger.debug("wait %fs", delay)
+                time.sleep(delay)
+            response = self.session.request(
+                method,
+                url,
+                **{["data", "json"][self.request_body_json]: params}
+                if method in ["POST", "PUT"]
+                else dict(params=params),
+                allow_redirects=False,
+            )
+            try:
+                # У этих лошков сервер не отдает Content-Length, а кривое API отдает пустые ответы, например, при отклике на вакансии
+                # 'Server': 'ddos-guard'
+                # ...
+                # 'Transfer-Encoding': 'chunked'
+                try:
+                    data = response.json()
+                except json.decoder.JSONDecodeError:
+                    if response.status_code in [201, 204]:
+                        data = {}
+                    else:
+                        raise
+            finally:
+                logger.debug(
+                    "%s %.88s %d",
+                    method,
+                    url + ("?" + urlencode(params) if params else ""),
+                    response.status_code,
+                )
+                self.previous_request_time = time.monotonic()
+        self.raise_for_status(response, data)
+        return data
+
+    get = partialmethod(request, "GET")
+    post = partialmethod(request, "POST")
+    put = partialmethod(request, "PUT")
+    delete = partialmethod(request, "DELETE")
+
+    def resolve_url(self, url: str) -> str:
+        return url if "://" in url else f"{self.base_url.rstrip('/')}/{url.lstrip('/')}"
+
+    @staticmethod
+    def raise_for_status(response: Response, data: dict) -> None:
+        match response.status_code:
+            case 400:
+                raise BadRequest(data)
+            case 403:
+                raise Forbidden(data)
+            case 404:
+                raise ResourceNotFound(data)
+            case 502:
+                raise BadGateaway(data)
+
+
+class AccessToken(TypedDict):
+    access_token: str
+    refresh_token: str
+    expires_in: int
+    token_type: Literal["bearer"]
+
+
+@dataclass
+class OAuthClient(BaseClient):
+    _: dataclasses.KW_ONLY
+    base_url: str = "https://hh.ru/oauth"
+    client_id: str = HHANDROID_CLIENT_ID
+    client_secret: str = HHANDROID_CLIENT_SECRET
+    state: str = ""
+    scope: str = ""
+    redirect_uri: str = ""
+
+    @property
+    def authorize_url(self) -> str:
+        params = dict(
+            client_id=self.client_id,
+            redirect_uri=self.redirect_uri,
+            response_type="code",
+            scope=self.scope,
+            state=self.state,
+        )
+        params_qs = urlencode({k: v for k, v in params.items() if v})
+        return self.resolve_url(f"/authorize?{params_qs}")
+
+    def authenticate(self, code: str) -> AccessToken:
+        params = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "code": code,
+            "grant_type": "authorization_code",
+        }
+        return self.request("POST", "/token", params)
+
+    def refresh_access(self, refresh_token: str) -> AccessToken:
+        # refresh_token можно использовать только один раз и только по истечению срока действия access_token.
+        return self.request(
+            "POST",
+            "/token",
+            {"grant_type": "refresh_token", "refresh_token": refresh_token},
+        )
+
+
+@dataclass
+class ApiClient(BaseClient):
+    _: dataclasses.KW_ONLY
+    access_token: str | None = None
+    refresh_token: str | None = None
+    base_url: str = "https://api.hh.ru/"
+    # request_body_json: bool = True
+    oauth_client: OAuthClient | None = None
+
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        self.oauth_client = self.oauth_client or OAuthClient(session=self.session)
+
+    def additional_headers(self) -> dict[str, str]:
+        return {"Authorization": f"Bearer {self.access_token}"}
+
+    def refresh_access(self) -> None:
+        tok = self.oauth_client.refresh_access(self.refresh_token)
+        self.access_token, self.refresh_access = (
+            tok["access_token"],
+            tok["refresh_token"],
+        )