# PharmaRadar
[](https://github.com/bartekmp/pharmaradar/actions/workflows/test.yml)
[](https://github.com/bartekmp/pharmaradar/actions/workflows/ci.yml)
[](https://pypi.org/project/pharmaradar/)
[](LICENSE)
Python package for searching and managing pharmacy medicine availability from [KtoMaLek.pl](https://ktomalek.pl).
## Requirements
Pharmaradar requires `chromium-browser`, `chromium-chromedriver` and `xvfb` to run, as the prerequisites for Selenium used to scrape the data from the KtoMaLek.pl page, as they do not provide an open API to get the data easily.
## Installation
```bash
pip install pharmaradar
```
## Usage
To work with searches use the `Medicine` object, which represents a search query including all required details about what you're looking for.
If you'd like to find nearest pharmacies, that have at least low availability of Euthyrox N 50 medicine, nearby the location like Złota street in Warsaw and the max radius of 10 kilometers, create it like this:
```python
import pharmaradar
medicine = pharmaradar.Medicine(
name="Euthyrox N 50",
dosage="50 mcg",
location="Warszawa, Złota",
radius_km=10.0,
min_availability=AvailabilityLevel.LOW,
)
```
Now create an instance of `MedicineFinder` class:
```python
finder = pharmaradar.MedicineFinder()
```
Then test if the connection to KtoMaLek.pl is possible and search for given medicine:
```python
if finder.test_connection():
pharmacies = finder.search_medicine(medicine)
```
If the search was successful, the `pharmacies` will contain a list of `PharmacyInfo` objects, with all important data found on the page:
```python
for pharmacy in pharmacies:
print(f"Pharmacy Name: {pharmacy.name}")
print(f"Address: {pharmacy.address}")
print(f"Availability: {pharmacy.availability}")
if pharmacy.price_full:
print(f"Price: {pharmacy.price_full} zł")
if pharmacy.distance_km:
print(f"Distance: {pharmacy.distance_km} km")
if pharmacy.reservation_url:
print(f"Reservation URL: {pharmacy.reservation_url}")
```
## Medicine watchdog
`MedicineWatchdog` is a class useful in async and continuous tasks. It implements certain methods, like `add_medicine`, `update_medicine`, `remove_medicine`, `get_medicine`, etc. that interact with the database layer, which is responsible for operating on the actual database. It can be used to create an automated bot, which periodically will retrieve the medicine quieries using `get_all_medicines` method, and then will perform searching and notifying.
```python
import sqlite3
from time import sleep
sql_db_client = SqliteInterface("my_database.db")
watchdog = pharmaradar.MedicineWatchdog(db_client)
while True:
all_medicines: list[Medicine] = watchdog.get_all_medicines()
for medicine in all_medicines:
print(f"Medicine: {medicine.name}")
found_pharmacies_for_medicine: list[PharmacyInfo] = await watchdog.search_medicine(medicine)
if found_pharmacies_for_medicine:
print(f"Found {len(found_pharmacies_for_medicine)}")
for p in found_pharmacies_for_medicine:
print(str(p))
else:
print(f"Medicine not available in pharmacies located in {medicine.distance_km} kilometer distance")
sleep(60) # 1 minute
```
### Database interface
The database interface instance passed to `MedicineWatchdog` must implement `MedicineDatabaseInterface`, which is basically a CRUD interface. The watchdog object will use this interface to interact with the data in the table. Example for an implementation for `sqlite` database:
```python
from pharmaradar import Medicine, MedicineDatabaseInterface
class SqliteInterface(MedicineDatabaseInterface):
def __init__(self, db_file_path: str):
self.conn = sqlite3.connect(db_path)
self.cur = self.conn.cursor()
def _parse_row_to_medicine(self, row: tuple) -> Medicine:
"""Convert a database row to a Medicine object."""
medicine_data = {
"id": row[0],
"name": row[1],
"dosage": row[2],
"amount": row[3],
"location": row[4],
"radius_km": row[5],
"max_price": row[6],
"min_availability": row[7],
"title": row[8],
"created_at": datetime.datetime.fromisoformat(row[9]) if row[9] else None,
"last_search_at": datetime.datetime.fromisoformat(row[10]) if row[10] else None,
"active": row[11], # Default to True for existing records
}
return Medicine(**medicine_data)
def get_medicine(self, medicine_id: int) -> Medicine | None:
row = self.cur.execute("SELECT * FROM medicine WHERE id = ?", (medicine_id,)).fetchone()
if row is None:
return None
return self._parse_row_to_medicine(row)
def get_medicines(self) -> list[Medicine]:
rows = self.cur.execute("SELECT * FROM medicine").fetchall()
medicines = []
for medicine_row in res:
medicines.append(self._parse_row_to_medicine(medicine_row))
return medicines
def remove_medicine(self, medicine_id: int) -> bool:
with self.conn:
res = self.cur.execute("DELETE FROM medicine WHERE id = (?)", (medicine_id,))
return res.rowcount > 0
def save_medicine(self, medicine: Medicine) -> int:
with self.conn:
self.cur.execute(
"INSERT INTO medicine VALUES (NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
medicine.name,
medicine.dosage,
medicine.amount,
medicine.location,
medicine.radius_km,
medicine.max_price,
medicine.min_availability.value,
medicine.title,
medicine.created_at.isoformat() if medicine.created_at else None,
medicine.last_search_at.isoformat() if medicine.last_search_at else None,
medicine.active,
),
)
return self.cur.lastrowid or 0
def update_medicine(
self,
medicine_id: int,
*,
name: str | None = None,
dosage: str | None = None,
amount: str | None = None,
location: str | None = None,
radius_km: float | None = None,
max_price: float | None = None,
min_availability: str | None = None,
title: str | None = None,
last_search_at: datetime.datetime | None = None,
active: bool | None = None,
) -> bool:
sql = []
values = []
if name is not None:
sql.append("name = ?")
values.append(name)
if dosage is not None:
sql.append("dosage = ?")
values.append(dosage)
if amount is not None:
sql.append("amount = ?")
values.append(amount)
if location is not None:
sql.append("location = ?")
values.append(location)
if radius_km is not None:
sql.append("radius_km = ?")
values.append(radius_km)
if max_price is not None:
sql.append("max_price = ?")
values.append(max_price)
if min_availability is not None:
sql.append("min_availability = ?")
values.append(min_availability)
if title is not None:
sql.append("title = ?")
values.append(title)
if last_search_at is not None:
sql.append("last_search_at = ?")
values.append(last_search_at.isoformat())
if active is not None:
sql.append("active = ?")
values.append(active)
values.append(medicine_id)
sql = f"UPDATE medicine SET {', '.join(sql)} WHERE id = ?"
with self.conn:
result = self.cur.execute(sql, values)
return result.rowcount > 0
```
Currently, the database itself must define the `medicine` table, declared as follows:
```sql
medicine(
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
dosage TEXT,
amount TEXT,
location TEXT NOT NULL,
radius_km REAL DEFAULT 10,
max_price REAL,
min_availability TEXT DEFAULT 'low',
title TEXT,
created_at TEXT,
last_search_at TEXT,
active BOOLEAN DEFAULT 1
)
```
## License
MIT License
Raw data
{
"_id": null,
"home_page": null,
"name": "pharmaradar",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "ktomalek, pharmaradar, web scraping, selenium",
"author": "bartekmp",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/39/e7/a59ea82a09e598d40e7c47dce24cf85bd7febe18f8fd562888008eac44b4/pharmaradar-1.1.2.tar.gz",
"platform": null,
"description": "# PharmaRadar\n\n[](https://github.com/bartekmp/pharmaradar/actions/workflows/test.yml)\n[](https://github.com/bartekmp/pharmaradar/actions/workflows/ci.yml)\n[](https://pypi.org/project/pharmaradar/)\n[](LICENSE)\n\nPython package for searching and managing pharmacy medicine availability from [KtoMaLek.pl](https://ktomalek.pl).\n\n## Requirements\nPharmaradar requires `chromium-browser`, `chromium-chromedriver` and `xvfb` to run, as the prerequisites for Selenium used to scrape the data from the KtoMaLek.pl page, as they do not provide an open API to get the data easily.\n\n## Installation\n\n```bash\npip install pharmaradar\n```\n\n## Usage\n\nTo work with searches use the `Medicine` object, which represents a search query including all required details about what you're looking for.\nIf you'd like to find nearest pharmacies, that have at least low availability of Euthyrox N 50 medicine, nearby the location like Z\u0142ota street in Warsaw and the max radius of 10 kilometers, create it like this:\n```python\nimport pharmaradar\n\nmedicine = pharmaradar.Medicine(\n name=\"Euthyrox N 50\",\n dosage=\"50 mcg\",\n location=\"Warszawa, Z\u0142ota\",\n radius_km=10.0,\n min_availability=AvailabilityLevel.LOW,\n )\n```\n\nNow create an instance of `MedicineFinder` class:\n```python\nfinder = pharmaradar.MedicineFinder()\n```\n\nThen test if the connection to KtoMaLek.pl is possible and search for given medicine:\n```python\nif finder.test_connection():\n pharmacies = finder.search_medicine(medicine)\n```\n\nIf the search was successful, the `pharmacies` will contain a list of `PharmacyInfo` objects, with all important data found on the page:\n```python\nfor pharmacy in pharmacies:\n print(f\"Pharmacy Name: {pharmacy.name}\")\n print(f\"Address: {pharmacy.address}\")\n print(f\"Availability: {pharmacy.availability}\")\n if pharmacy.price_full:\n print(f\"Price: {pharmacy.price_full} z\u0142\")\n if pharmacy.distance_km:\n print(f\"Distance: {pharmacy.distance_km} km\")\n if pharmacy.reservation_url:\n print(f\"Reservation URL: {pharmacy.reservation_url}\")\n```\n\n## Medicine watchdog\n\n`MedicineWatchdog` is a class useful in async and continuous tasks. It implements certain methods, like `add_medicine`, `update_medicine`, `remove_medicine`, `get_medicine`, etc. that interact with the database layer, which is responsible for operating on the actual database. It can be used to create an automated bot, which periodically will retrieve the medicine quieries using `get_all_medicines` method, and then will perform searching and notifying.\n```python\nimport sqlite3\nfrom time import sleep\n\nsql_db_client = SqliteInterface(\"my_database.db\")\nwatchdog = pharmaradar.MedicineWatchdog(db_client)\n\nwhile True:\n\n all_medicines: list[Medicine] = watchdog.get_all_medicines()\n for medicine in all_medicines:\n\n print(f\"Medicine: {medicine.name}\")\n\n found_pharmacies_for_medicine: list[PharmacyInfo] = await watchdog.search_medicine(medicine)\n\n if found_pharmacies_for_medicine:\n\n print(f\"Found {len(found_pharmacies_for_medicine)}\")\n\n for p in found_pharmacies_for_medicine:\n print(str(p))\n else:\n print(f\"Medicine not available in pharmacies located in {medicine.distance_km} kilometer distance\")\n\n sleep(60) # 1 minute\n```\n\n### Database interface\nThe database interface instance passed to `MedicineWatchdog` must implement `MedicineDatabaseInterface`, which is basically a CRUD interface. The watchdog object will use this interface to interact with the data in the table. Example for an implementation for `sqlite` database:\n\n```python\nfrom pharmaradar import Medicine, MedicineDatabaseInterface\n\nclass SqliteInterface(MedicineDatabaseInterface):\n def __init__(self, db_file_path: str):\n self.conn = sqlite3.connect(db_path)\n self.cur = self.conn.cursor()\n\n def _parse_row_to_medicine(self, row: tuple) -> Medicine:\n \"\"\"Convert a database row to a Medicine object.\"\"\"\n medicine_data = {\n \"id\": row[0],\n \"name\": row[1],\n \"dosage\": row[2],\n \"amount\": row[3],\n \"location\": row[4],\n \"radius_km\": row[5],\n \"max_price\": row[6],\n \"min_availability\": row[7],\n \"title\": row[8],\n \"created_at\": datetime.datetime.fromisoformat(row[9]) if row[9] else None,\n \"last_search_at\": datetime.datetime.fromisoformat(row[10]) if row[10] else None,\n \"active\": row[11], # Default to True for existing records\n }\n return Medicine(**medicine_data)\n\n def get_medicine(self, medicine_id: int) -> Medicine | None:\n row = self.cur.execute(\"SELECT * FROM medicine WHERE id = ?\", (medicine_id,)).fetchone()\n if row is None:\n return None\n return self._parse_row_to_medicine(row)\n\n def get_medicines(self) -> list[Medicine]:\n rows = self.cur.execute(\"SELECT * FROM medicine\").fetchall()\n medicines = []\n for medicine_row in res:\n medicines.append(self._parse_row_to_medicine(medicine_row))\n return medicines\n\n def remove_medicine(self, medicine_id: int) -> bool:\n with self.conn:\n res = self.cur.execute(\"DELETE FROM medicine WHERE id = (?)\", (medicine_id,))\n return res.rowcount > 0\n\n def save_medicine(self, medicine: Medicine) -> int:\n with self.conn:\n self.cur.execute(\n \"INSERT INTO medicine VALUES (NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)\",\n (\n medicine.name,\n medicine.dosage,\n medicine.amount,\n medicine.location,\n medicine.radius_km,\n medicine.max_price,\n medicine.min_availability.value,\n medicine.title,\n medicine.created_at.isoformat() if medicine.created_at else None,\n medicine.last_search_at.isoformat() if medicine.last_search_at else None,\n medicine.active,\n ),\n )\n return self.cur.lastrowid or 0\n\n def update_medicine(\n self,\n medicine_id: int,\n *,\n name: str | None = None,\n dosage: str | None = None,\n amount: str | None = None,\n location: str | None = None,\n radius_km: float | None = None,\n max_price: float | None = None,\n min_availability: str | None = None,\n title: str | None = None,\n last_search_at: datetime.datetime | None = None,\n active: bool | None = None,\n ) -> bool:\n sql = []\n values = []\n if name is not None:\n sql.append(\"name = ?\")\n values.append(name)\n if dosage is not None:\n sql.append(\"dosage = ?\")\n values.append(dosage)\n if amount is not None:\n sql.append(\"amount = ?\")\n values.append(amount)\n if location is not None:\n sql.append(\"location = ?\")\n values.append(location)\n if radius_km is not None:\n sql.append(\"radius_km = ?\")\n values.append(radius_km)\n if max_price is not None:\n sql.append(\"max_price = ?\")\n values.append(max_price)\n if min_availability is not None:\n sql.append(\"min_availability = ?\")\n values.append(min_availability)\n if title is not None:\n sql.append(\"title = ?\")\n values.append(title)\n if last_search_at is not None:\n sql.append(\"last_search_at = ?\")\n values.append(last_search_at.isoformat())\n if active is not None:\n sql.append(\"active = ?\")\n values.append(active)\n\n values.append(medicine_id)\n sql = f\"UPDATE medicine SET {', '.join(sql)} WHERE id = ?\"\n\n with self.conn:\n result = self.cur.execute(sql, values)\n return result.rowcount > 0 \n \n```\n\nCurrently, the database itself must define the `medicine` table, declared as follows:\n```sql\nmedicine(\n id INTEGER PRIMARY KEY,\n name TEXT NOT NULL,\n dosage TEXT,\n amount TEXT,\n location TEXT NOT NULL,\n radius_km REAL DEFAULT 10,\n max_price REAL,\n min_availability TEXT DEFAULT 'low',\n title TEXT,\n created_at TEXT,\n last_search_at TEXT,\n active BOOLEAN DEFAULT 1\n )\n```\n\n## License\n\nMIT License\n",
"bugtrack_url": null,
"license": null,
"summary": "Python scraping package for KtoMaLek.pl website to monitor drug availability.",
"version": "1.1.2",
"project_urls": null,
"split_keywords": [
"ktomalek",
" pharmaradar",
" web scraping",
" selenium"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "768626f4716458c7b349688fc4bb0fcbf57eb5f708d5a10790eb52d613e233ea",
"md5": "89d4dfdee5a1baeed5810dd5705dd158",
"sha256": "074b281373cdc05ca5799498c4bfa4420098d49dd38a387992a37b9f35019a89"
},
"downloads": -1,
"filename": "pharmaradar-1.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "89d4dfdee5a1baeed5810dd5705dd158",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 35823,
"upload_time": "2025-09-03T16:25:37",
"upload_time_iso_8601": "2025-09-03T16:25:37.254711Z",
"url": "https://files.pythonhosted.org/packages/76/86/26f4716458c7b349688fc4bb0fcbf57eb5f708d5a10790eb52d613e233ea/pharmaradar-1.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "39e7a59ea82a09e598d40e7c47dce24cf85bd7febe18f8fd562888008eac44b4",
"md5": "4d77a406aea7f077b0bf0f49a18559f7",
"sha256": "ad6fd0d44dba39b50f28ff2e88a398e1b6f98716fa8a64923896c3de184574b5"
},
"downloads": -1,
"filename": "pharmaradar-1.1.2.tar.gz",
"has_sig": false,
"md5_digest": "4d77a406aea7f077b0bf0f49a18559f7",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 48458,
"upload_time": "2025-09-03T16:25:38",
"upload_time_iso_8601": "2025-09-03T16:25:38.674752Z",
"url": "https://files.pythonhosted.org/packages/39/e7/a59ea82a09e598d40e7c47dce24cf85bd7febe18f8fd562888008eac44b4/pharmaradar-1.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-03 16:25:38",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "pharmaradar"
}
JSON