diff --git a/evoting_refactored (1)/evoting_data.json b/evoting_refactored (1)/evoting_data.json new file mode 100644 index 0000000..ef78d42 --- /dev/null +++ b/evoting_refactored (1)/evoting_data.json @@ -0,0 +1,34 @@ +{ + "candidates": {}, + "candidate_id_counter": 1, + "voting_stations": {}, + "station_id_counter": 1, + "polls": {}, + "poll_id_counter": 1, + "positions": {}, + "position_id_counter": 1, + "voters": {}, + "voter_id_counter": 1, + "admins": { + "1": { + "id": 1, + "username": "admin", + "password": "240be518fabd2724ddb6f04eeb1da5967448d7e831c08c8fa822809f74c720a9", + "full_name": "System Administrator", + "email": "admin@evote.com", + "role": "super_admin", + "created_at": "2026-03-12 23:39:29.971270", + "is_active": true + } + }, + "admin_id_counter": 2, + "votes": [], + "audit_log": [ + { + "timestamp": "2026-03-12 23:39:53.962079", + "action": "LOGIN_FAILED", + "user": "rfe244", + "details": "Invalid voter card number or password" + } + ] +} \ No newline at end of file diff --git a/evoting_refactored (1)/evoting_refactored/evoting/README.md b/evoting_refactored (1)/evoting_refactored/evoting/README.md new file mode 100644 index 0000000..7d0bc51 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/README.md @@ -0,0 +1,125 @@ +# National E-Voting System — Refactored + +**Course:** Software Construction +**Year:** 3, Semester 2 — Easter 2026 +**Program:** BSc Computer Science, Uganda Christian University + +--- + +## Overview + +This project is a full refactor of a monolithic National E-Voting console application into a clean, modular, object-oriented Python codebase. The application behaviour is identical to the original — same menus, same prompts, same outputs — but the internal architecture has been completely restructured to follow four key software engineering principles: + +1. Modular Design +2. Object-Oriented Design +3. Separation of Concerns +4. Clean Code + +--- + +## How to Run + +```bash +cd evoting +python3 main.py +``` + +Default admin credentials: `admin` / `admin123` + +--- + +## Project Structure + +``` +evoting/ +├── main.py # Entry point only — wires everything together +├── config.py # Application-wide constants +│ +├── ui/ +│ ├── colors.py # ANSI color codes and theme constants +│ ├── components.py # Reusable display functions (header, table, menu) +│ └── prompts.py # Input helpers (prompt, masked_input, pause) +│ +├── models/ +│ ├── candidate.py # Candidate data class +│ ├── voter.py # Voter data class +│ ├── admin.py # Admin data class +│ ├── poll.py # Poll, PollPosition, and Position classes +│ ├── station.py # VotingStation data class +│ └── vote.py # Vote data class +│ +├── services/ +│ ├── auth_service.py # Login validation and session management +│ ├── audit_service.py # Audit log recording +│ ├── candidate_service.py # Candidate CRUD and eligibility checks +│ ├── voter_service.py # Voter registration, verification, password change +│ ├── station_service.py # Voting station CRUD +│ ├── poll_service.py # Poll and Position lifecycle management +│ ├── vote_service.py # Ballot casting with duplicate prevention +│ ├── admin_service.py # Admin account creation and deactivation +│ └── results_service.py # Vote tallying, turnout, and statistics +│ +├── data/ +│ └── storage.py # Central store, JSON save/load, data seeding +│ +└── views/ + ├── auth_view.py # Login menu and voter self-registration screen + ├── admin_view.py # All admin-facing menus and screens + └── voter_view.py # All voter-facing menus and screens +``` + +--- + +## Design Decisions + +### 1. Modular Design + +The original monolith had all code — constants, global state, UI, and logic — in a single 700+ line file. We split this into **29 focused files** across 5 packages, each with a single clear responsibility. The `config.py` file centralises all constants, eliminating magic numbers and hard-coded strings scattered through the original. + +### 2. Object-Oriented Design + +The original used plain Python dictionaries to represent every entity. We replaced these with proper classes in the `models/` package: + +- Each model (`Candidate`, `Voter`, `Admin`, `Poll`, `VotingStation`, `Vote`) encapsulates its own data and provides `to_dict()` / `from_dict()` methods for serialisation. +- The `Store` class in `data/storage.py` encapsulates all application state and persistence logic. +- The `AuthService` class manages the user session with clear `login_admin()`, `login_voter()`, and `logout()` methods. +- View classes (`AdminView`, `VoterView`, `AuthView`) encapsulate all UI behaviour per role. +- Service classes receive the store via constructor injection (dependency injection), making each testable in isolation. + +### 3. Separation of Concerns + +The architecture enforces three strict layers: + +| Layer | Package | Responsibility | +|---|---|---| +| Presentation | `views/` | Reads input, prints output. Never contains business logic. | +| Business Logic | `services/` | Validates rules, computes results. Never prints or reads input directly. | +| Data | `models/` + `data/` | Defines data shape and handles persistence. No logic, no UI. | + +A key example is `cast_vote()`. In the original, one function handled input collection, duplicate checking, vote recording, voter state update, and file saving. In the refactored version, `VoterView.cast_vote()` handles only input/output, and delegates to `VoteService.cast_ballot()` for all logic. The service raises a `ValueError` on any violation; the view catches it and displays the message. + +### 4. Clean Code + +- All functions and methods have a single clear purpose. +- Meaningful names replace cryptic abbreviations (e.g. `tc`, `ac`, `vc`). +- `ValueError` exceptions replace inline `error()` + `return` patterns in business logic. +- Magic values like `25`, `75`, `18`, and `6` are constants in `config.py`. +- No duplicated code — shared display logic lives in `ui/components.py` and is reused across all views. +- A singleton `store` and `session` are injected into views and services, avoiding global variables. + +--- + +## Features Preserved + +All original features work identically after refactoring: + +- Candidate CRUD with age, education, and criminal record eligibility checks +- Voting station management +- Position and poll lifecycle (draft → open → closed) +- Voter registration and verification +- Ballot casting with duplicate prevention +- Result tallying with ASCII bar charts and turnout statistics +- Station-wise result breakdowns +- Role-based access for 4 admin types and voters +- Audit log with filtering +- JSON persistence diff --git a/evoting_refactored (1)/evoting_refactored/evoting/__pycache__/config.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/__pycache__/config.cpython-313.pyc new file mode 100644 index 0000000..22529b1 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/__pycache__/config.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/config.py b/evoting_refactored (1)/evoting_refactored/evoting/config.py new file mode 100644 index 0000000..9c04f46 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/config.py @@ -0,0 +1,37 @@ +""" +Application-wide constants and configuration. +""" + +MIN_CANDIDATE_AGE = 25 +MAX_CANDIDATE_AGE = 75 +MIN_VOTER_AGE = 18 +MIN_PASSWORD_LENGTH = 6 + +REQUIRED_EDUCATION_LEVELS = [ + "Bachelor's Degree", + "Master's Degree", + "PhD", + "Doctorate", +] + +ADMIN_ROLES = { + "1": "super_admin", + "2": "election_officer", + "3": "station_manager", + "4": "auditor", +} + +ADMIN_ROLE_LABELS = { + "super_admin": "Full access", + "election_officer": "Manage polls and candidates", + "station_manager": "Manage stations and verify voters", + "auditor": "Read-only access", +} + +ELECTION_TYPES = ["General", "Primary", "By-election", "Referendum"] +POSITION_LEVELS = ["national", "regional", "local"] +DATA_FILE = "evoting_data.json" +DEFAULT_ADMIN_USERNAME = "admin" +DEFAULT_ADMIN_PASSWORD = "admin123" +DEFAULT_ADMIN_FULL_NAME = "System Administrator" +DEFAULT_ADMIN_EMAIL = "admin@evote.com" diff --git a/evoting_refactored (1)/evoting_refactored/evoting/data/__init__.py b/evoting_refactored (1)/evoting_refactored/evoting/data/__init__.py new file mode 100644 index 0000000..d6f6516 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/data/__init__.py @@ -0,0 +1 @@ +# data package diff --git a/evoting_refactored (1)/evoting_refactored/evoting/data/__pycache__/__init__.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/data/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..b0a6609 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/data/__pycache__/__init__.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/data/__pycache__/storage.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/data/__pycache__/storage.cpython-313.pyc new file mode 100644 index 0000000..1382095 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/data/__pycache__/storage.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/data/storage.py b/evoting_refactored (1)/evoting_refactored/evoting/data/storage.py new file mode 100644 index 0000000..f92b445 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/data/storage.py @@ -0,0 +1,138 @@ +""" +Data storage layer — holds all application state and handles +JSON persistence. This is the single source of truth for all data. +""" + +import datetime +import json +import os + +from config import DATA_FILE, DEFAULT_ADMIN_USERNAME, DEFAULT_ADMIN_PASSWORD, DEFAULT_ADMIN_FULL_NAME, DEFAULT_ADMIN_EMAIL +from models.admin import Admin +from models.candidate import Candidate +from models.voter import Voter +from models.station import VotingStation +from models.poll import Poll, Position +from models.vote import Vote +from services.auth_service import hash_password + + +class Store: + """Central data store for the entire application.""" + + def __init__(self): + self.candidates: dict[int, Candidate] = {} + self.candidate_id_counter: int = 1 + + self.voting_stations: dict[int, VotingStation] = {} + self.station_id_counter: int = 1 + + self.polls: dict[int, Poll] = {} + self.poll_id_counter: int = 1 + + self.positions: dict[int, Position] = {} + self.position_id_counter: int = 1 + + self.voters: dict[int, Voter] = {} + self.voter_id_counter: int = 1 + + self.admins: dict[int, Admin] = {} + self.admin_id_counter: int = 1 + + self.votes: list[Vote] = [] + self.audit_log: list[dict] = [] + + self._seed_default_admin() + + def _seed_default_admin(self): + """Create the default super admin account on first run.""" + self.admins[1] = Admin( + id=1, + username=DEFAULT_ADMIN_USERNAME, + password=hash_password(DEFAULT_ADMIN_PASSWORD), + full_name=DEFAULT_ADMIN_FULL_NAME, + email=DEFAULT_ADMIN_EMAIL, + role="super_admin", + created_at=str(datetime.datetime.now()), + is_active=True, + ) + self.admin_id_counter = 2 + + def save(self): + """Serialize all state to JSON and write to disk.""" + data = { + "candidates": {k: v.to_dict() for k, v in self.candidates.items()}, + "candidate_id_counter": self.candidate_id_counter, + "voting_stations": {k: v.to_dict() for k, v in self.voting_stations.items()}, + "station_id_counter": self.station_id_counter, + "polls": {k: v.to_dict() for k, v in self.polls.items()}, + "poll_id_counter": self.poll_id_counter, + "positions": {k: v.to_dict() for k, v in self.positions.items()}, + "position_id_counter": self.position_id_counter, + "voters": {k: v.to_dict() for k, v in self.voters.items()}, + "voter_id_counter": self.voter_id_counter, + "admins": {k: v.to_dict() for k, v in self.admins.items()}, + "admin_id_counter": self.admin_id_counter, + "votes": [v.to_dict() for v in self.votes], + "audit_log": self.audit_log, + } + try: + with open(DATA_FILE, "w") as f: + json.dump(data, f, indent=2) + except Exception as e: + print(f" Error saving data: {e}") + + def load(self): + """Load and deserialize state from JSON file if it exists.""" + if not os.path.exists(DATA_FILE): + return + + try: + with open(DATA_FILE, "r") as f: + data = json.load(f) + + self.candidates = { + int(k): Candidate.from_dict(v) + for k, v in data.get("candidates", {}).items() + } + self.candidate_id_counter = data.get("candidate_id_counter", 1) + + self.voting_stations = { + int(k): VotingStation.from_dict(v) + for k, v in data.get("voting_stations", {}).items() + } + self.station_id_counter = data.get("station_id_counter", 1) + + self.polls = { + int(k): Poll.from_dict(v) + for k, v in data.get("polls", {}).items() + } + self.poll_id_counter = data.get("poll_id_counter", 1) + + self.positions = { + int(k): Position.from_dict(v) + for k, v in data.get("positions", {}).items() + } + self.position_id_counter = data.get("position_id_counter", 1) + + self.voters = { + int(k): Voter.from_dict(v) + for k, v in data.get("voters", {}).items() + } + self.voter_id_counter = data.get("voter_id_counter", 1) + + self.admins = { + int(k): Admin.from_dict(v) + for k, v in data.get("admins", {}).items() + } + self.admin_id_counter = data.get("admin_id_counter", 1) + + self.votes = [Vote.from_dict(v) for v in data.get("votes", [])] + self.audit_log = data.get("audit_log", []) + + except Exception as e: + print(f" Error loading data: {e}") + + +# Singleton store instance used across the application +store = Store() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/main.py b/evoting_refactored (1)/evoting_refactored/evoting/main.py new file mode 100644 index 0000000..87bcb96 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/main.py @@ -0,0 +1,63 @@ +""" +main.py — Application entry point. + +Wires together the data store, services, and views, +then starts the main application loop. +""" + +import time + +from ui.colors import THEME_LOGIN, RESET +from ui.prompts import clear_screen +from data.storage import store +from services.auth_service import session +from services.candidate_service import CandidateService +from services.voter_service import VoterService +from services.station_service import StationService +from services.poll_service import PollService, PositionService +from services.vote_service import VoteService +from services.admin_service import AdminService +from services.results_service import ResultsService +from views.auth_view import AuthView +from views.admin_view import AdminView +from views.voter_view import VoterView + + +def build_services(store): + """Instantiate all services, injecting the shared store.""" + return { + "candidate": CandidateService(store), + "voter": VoterService(store), + "station": StationService(store), + "position": PositionService(store), + "poll": PollService(store), + "vote": VoteService(store), + "admin": AdminService(store), + "results": ResultsService(store), + } + + +def main(): + print(f"\n {THEME_LOGIN}Loading E-Voting System...{RESET}") + store.load() + time.sleep(1) + + services = build_services(store) + auth_view = AuthView(store, session, services["voter"]) + admin_view = AdminView(store, session, services) + voter_view = VoterView(store, session, services) + + while True: + clear_screen() + logged_in = auth_view.show_login_menu() + + if logged_in: + if session.current_role == "admin": + admin_view.run() + elif session.current_role == "voter": + voter_view.run() + session.logout() + + +if __name__ == "__main__": + main() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__init__.py b/evoting_refactored (1)/evoting_refactored/evoting/models/__init__.py new file mode 100644 index 0000000..3c159a6 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/__init__.py @@ -0,0 +1 @@ +# models package diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/__init__.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..59313e3 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/__init__.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/admin.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/admin.cpython-313.pyc new file mode 100644 index 0000000..c22ed36 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/admin.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/candidate.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/candidate.cpython-313.pyc new file mode 100644 index 0000000..e6ec1c0 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/candidate.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/poll.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/poll.cpython-313.pyc new file mode 100644 index 0000000..d283f93 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/poll.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/station.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/station.cpython-313.pyc new file mode 100644 index 0000000..11d8729 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/station.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/vote.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/vote.cpython-313.pyc new file mode 100644 index 0000000..468213e Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/vote.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/voter.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/voter.cpython-313.pyc new file mode 100644 index 0000000..71f06eb Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/models/__pycache__/voter.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/admin.py b/evoting_refactored (1)/evoting_refactored/evoting/models/admin.py new file mode 100644 index 0000000..410be52 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/admin.py @@ -0,0 +1,26 @@ +""" +Admin model — represents an administrative user account. +No business logic; only data structure and serialization. +""" + + +class Admin: + def __init__( + self, id, username, password, full_name, email, + role, created_at, is_active, + ): + self.id = id + self.username = username + self.password = password + self.full_name = full_name + self.email = email + self.role = role + self.created_at = created_at + self.is_active = is_active + + def to_dict(self): + return self.__dict__.copy() + + @classmethod + def from_dict(cls, data): + return cls(**data) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/candidate.py b/evoting_refactored (1)/evoting_refactored/evoting/models/candidate.py new file mode 100644 index 0000000..31a1786 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/candidate.py @@ -0,0 +1,40 @@ +""" +Candidate model — represents a candidate's data and identity. +No business logic; only data structure and serialization. +""" + +import datetime + + +class Candidate: + def __init__( + self, id, full_name, national_id, date_of_birth, age, gender, + education, party, manifesto, address, phone, email, + has_criminal_record, years_experience, is_active, is_approved, + created_at, created_by, + ): + self.id = id + self.full_name = full_name + self.national_id = national_id + self.date_of_birth = date_of_birth + self.age = age + self.gender = gender + self.education = education + self.party = party + self.manifesto = manifesto + self.address = address + self.phone = phone + self.email = email + self.has_criminal_record = has_criminal_record + self.years_experience = years_experience + self.is_active = is_active + self.is_approved = is_approved + self.created_at = created_at + self.created_by = created_by + + def to_dict(self): + return self.__dict__.copy() + + @classmethod + def from_dict(cls, data): + return cls(**data) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/poll.py b/evoting_refactored (1)/evoting_refactored/evoting/models/poll.py new file mode 100644 index 0000000..9f50d73 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/poll.py @@ -0,0 +1,111 @@ +""" +Poll and Position models — represent elections and the positions within them. +No business logic; only data structure and serialization. +""" + + +class PollPosition: + """A position (e.g. President) within a specific poll.""" + + def __init__(self, position_id, position_title, candidate_ids, max_winners): + self.position_id = position_id + self.position_title = position_title + self.candidate_ids = candidate_ids if candidate_ids is not None else [] + self.max_winners = max_winners + + def to_dict(self): + return { + "position_id": self.position_id, + "position_title": self.position_title, + "candidate_ids": self.candidate_ids, + "max_winners": self.max_winners, + } + + @classmethod + def from_dict(cls, data): + return cls( + position_id=data["position_id"], + position_title=data["position_title"], + candidate_ids=data.get("candidate_ids", []), + max_winners=data["max_winners"], + ) + + +class Poll: + """An election poll containing one or more positions.""" + + def __init__( + self, id, title, description, election_type, start_date, end_date, + positions, station_ids, status, total_votes_cast, created_at, created_by, + ): + self.id = id + self.title = title + self.description = description + self.election_type = election_type + self.start_date = start_date + self.end_date = end_date + self.positions = positions # list of PollPosition objects + self.station_ids = station_ids if station_ids is not None else [] + self.status = status + self.total_votes_cast = total_votes_cast + self.created_at = created_at + self.created_by = created_by + + def to_dict(self): + return { + "id": self.id, + "title": self.title, + "description": self.description, + "election_type": self.election_type, + "start_date": self.start_date, + "end_date": self.end_date, + "positions": [p.to_dict() for p in self.positions], + "station_ids": self.station_ids, + "status": self.status, + "total_votes_cast": self.total_votes_cast, + "created_at": self.created_at, + "created_by": self.created_by, + } + + @classmethod + def from_dict(cls, data): + positions = [PollPosition.from_dict(p) for p in data.get("positions", [])] + return cls( + id=data["id"], + title=data["title"], + description=data["description"], + election_type=data["election_type"], + start_date=data["start_date"], + end_date=data["end_date"], + positions=positions, + station_ids=data.get("station_ids", []), + status=data["status"], + total_votes_cast=data.get("total_votes_cast", 0), + created_at=data["created_at"], + created_by=data["created_by"], + ) + + +class Position: + """A reusable electoral position definition (e.g. President, Governor).""" + + def __init__( + self, id, title, description, level, max_winners, + min_candidate_age, is_active, created_at, created_by, + ): + self.id = id + self.title = title + self.description = description + self.level = level + self.max_winners = max_winners + self.min_candidate_age = min_candidate_age + self.is_active = is_active + self.created_at = created_at + self.created_by = created_by + + def to_dict(self): + return self.__dict__.copy() + + @classmethod + def from_dict(cls, data): + return cls(**data) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/station.py b/evoting_refactored (1)/evoting_refactored/evoting/models/station.py new file mode 100644 index 0000000..6e6146b --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/station.py @@ -0,0 +1,32 @@ +""" +VotingStation model — represents a physical voting station. +No business logic; only data structure and serialization. +""" + + +class VotingStation: + def __init__( + self, id, name, location, region, capacity, registered_voters, + supervisor, contact, opening_time, closing_time, + is_active, created_at, created_by, + ): + self.id = id + self.name = name + self.location = location + self.region = region + self.capacity = capacity + self.registered_voters = registered_voters + self.supervisor = supervisor + self.contact = contact + self.opening_time = opening_time + self.closing_time = closing_time + self.is_active = is_active + self.created_at = created_at + self.created_by = created_by + + def to_dict(self): + return self.__dict__.copy() + + @classmethod + def from_dict(cls, data): + return cls(**data) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/vote.py b/evoting_refactored (1)/evoting_refactored/evoting/models/vote.py new file mode 100644 index 0000000..f7159f2 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/vote.py @@ -0,0 +1,26 @@ +""" +Vote model — represents a single ballot entry for one position. +No business logic; only data structure and serialization. +""" + + +class Vote: + def __init__( + self, vote_id, poll_id, position_id, candidate_id, + voter_id, station_id, timestamp, abstained, + ): + self.vote_id = vote_id + self.poll_id = poll_id + self.position_id = position_id + self.candidate_id = candidate_id + self.voter_id = voter_id + self.station_id = station_id + self.timestamp = timestamp + self.abstained = abstained + + def to_dict(self): + return self.__dict__.copy() + + @classmethod + def from_dict(cls, data): + return cls(**data) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/models/voter.py b/evoting_refactored (1)/evoting_refactored/evoting/models/voter.py new file mode 100644 index 0000000..d109922 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/models/voter.py @@ -0,0 +1,36 @@ +""" +Voter model — represents a registered voter's data and identity. +No business logic; only data structure and serialization. +""" + + +class Voter: + def __init__( + self, id, full_name, national_id, date_of_birth, age, gender, + address, phone, email, password, voter_card_number, station_id, + is_verified, is_active, has_voted_in, registered_at, role="voter", + ): + self.id = id + self.full_name = full_name + self.national_id = national_id + self.date_of_birth = date_of_birth + self.age = age + self.gender = gender + self.address = address + self.phone = phone + self.email = email + self.password = password + self.voter_card_number = voter_card_number + self.station_id = station_id + self.is_verified = is_verified + self.is_active = is_active + self.has_voted_in = has_voted_in if has_voted_in is not None else [] + self.registered_at = registered_at + self.role = role + + def to_dict(self): + return self.__dict__.copy() + + @classmethod + def from_dict(cls, data): + return cls(**data) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__init__.py b/evoting_refactored (1)/evoting_refactored/evoting/services/__init__.py new file mode 100644 index 0000000..0274469 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/__init__.py @@ -0,0 +1 @@ +# services package diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/__init__.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..f4c5630 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/__init__.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/admin_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/admin_service.cpython-313.pyc new file mode 100644 index 0000000..f440a15 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/admin_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/audit_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/audit_service.cpython-313.pyc new file mode 100644 index 0000000..a5c2a8a Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/audit_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/auth_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/auth_service.cpython-313.pyc new file mode 100644 index 0000000..2a0c9f7 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/auth_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/candidate_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/candidate_service.cpython-313.pyc new file mode 100644 index 0000000..ccdf1a9 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/candidate_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/poll_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/poll_service.cpython-313.pyc new file mode 100644 index 0000000..a309ad0 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/poll_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/results_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/results_service.cpython-313.pyc new file mode 100644 index 0000000..3af9f72 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/results_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/station_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/station_service.cpython-313.pyc new file mode 100644 index 0000000..a555bbf Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/station_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/vote_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/vote_service.cpython-313.pyc new file mode 100644 index 0000000..54d7102 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/vote_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/voter_service.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/voter_service.cpython-313.pyc new file mode 100644 index 0000000..1649c87 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/services/__pycache__/voter_service.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/admin_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/admin_service.py new file mode 100644 index 0000000..59af6a7 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/admin_service.py @@ -0,0 +1,53 @@ +""" +Admin account service — handles creating and deactivating admin accounts. +No UI output. +""" + +import datetime + +from config import MIN_PASSWORD_LENGTH, ADMIN_ROLES +from models.admin import Admin +from services.auth_service import hash_password +from services.audit_service import log_action + + +class AdminService: + def __init__(self, store): + self.store = store + + def create(self, data: dict, created_by: str) -> Admin: + if not data["username"]: + raise ValueError("Username cannot be empty.") + for a in self.store.admins.values(): + if a.username == data["username"]: + raise ValueError("Username already exists.") + if len(data["password"]) < MIN_PASSWORD_LENGTH: + raise ValueError(f"Password must be at least {MIN_PASSWORD_LENGTH} characters.") + if data["role"] not in ADMIN_ROLES.values(): + raise ValueError("Invalid role.") + + admin = Admin( + id=self.store.admin_id_counter, + username=data["username"], + password=hash_password(data["password"]), + full_name=data["full_name"], + email=data["email"], + role=data["role"], + created_at=str(datetime.datetime.now()), + is_active=True, + ) + self.store.admins[admin.id] = admin + self.store.admin_id_counter += 1 + log_action(self.store, "CREATE_ADMIN", created_by, f"Created admin: {admin.username} (Role: {admin.role})") + self.store.save() + return admin + + def deactivate(self, admin_id: int, current_admin_id: int, deactivated_by: str): + admin = self.store.admins.get(admin_id) + if not admin: + raise ValueError("Admin not found.") + if admin_id == current_admin_id: + raise ValueError("Cannot deactivate your own account.") + admin.is_active = False + log_action(self.store, "DEACTIVATE_ADMIN", deactivated_by, f"Deactivated admin: {admin.username}") + self.store.save() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/audit_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/audit_service.py new file mode 100644 index 0000000..65e0475 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/audit_service.py @@ -0,0 +1,15 @@ +""" +Audit service — records all significant system actions to the audit log. +""" + +import datetime + + +def log_action(store, action: str, user: str, details: str): + """Append an audit entry to the store's audit log.""" + store.audit_log.append({ + "timestamp": str(datetime.datetime.now()), + "action": action, + "user": user, + "details": details, + }) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/auth_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/auth_service.py new file mode 100644 index 0000000..4897dcd --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/auth_service.py @@ -0,0 +1,64 @@ +""" +Authentication service — handles password hashing, login validation, +and the current user session. No UI output; returns results to callers. +""" + +import hashlib + + +def hash_password(password: str) -> str: + """Return the SHA-256 hex digest of a plaintext password.""" + return hashlib.sha256(password.encode()).hexdigest() + + +class AuthService: + """Manages the current user session and login verification.""" + + def __init__(self): + self.current_user = None + self.current_role: str | None = None + + def login_admin(self, username: str, password: str, admins: dict): + """ + Attempt admin login. Returns (success, message, admin_object). + """ + hashed = hash_password(password) + for admin in admins.values(): + if admin.username == username and admin.password == hashed: + if not admin.is_active: + return False, "Account deactivated", None + self.current_user = admin + self.current_role = "admin" + return True, f"Welcome, {admin.full_name}!", admin + return False, "Invalid credentials", None + + def login_voter(self, voter_card: str, password: str, voters: dict): + """ + Attempt voter login. Returns (success, message, voter_object). + """ + hashed = hash_password(password) + for voter in voters.values(): + if voter.voter_card_number == voter_card and voter.password == hashed: + if not voter.is_active: + return False, "Voter account deactivated", None + if not voter.is_verified: + return False, "Registration not yet verified. Contact an admin.", None + self.current_user = voter + self.current_role = "voter" + return True, f"Welcome, {voter.full_name}!", voter + return False, "Invalid voter card number or password", None + + def logout(self): + self.current_user = None + self.current_role = None + + def is_super_admin(self) -> bool: + return ( + self.current_user is not None + and self.current_role == "admin" + and self.current_user.role == "super_admin" + ) + + +# Singleton session used across the application +session = AuthService() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/candidate_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/candidate_service.py new file mode 100644 index 0000000..297a31e --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/candidate_service.py @@ -0,0 +1,111 @@ +""" +Candidate service — business logic for candidate CRUD and eligibility validation. +No UI output; raises ValueError or returns results to callers. +""" + +import datetime + +from config import MIN_CANDIDATE_AGE, MAX_CANDIDATE_AGE, REQUIRED_EDUCATION_LEVELS +from models.candidate import Candidate +from services.audit_service import log_action + + +def calculate_age(dob_str: str) -> int: + dob = datetime.datetime.strptime(dob_str, "%Y-%m-%d") + return (datetime.datetime.now() - dob).days // 365 + + +class CandidateService: + def __init__(self, store): + self.store = store + + def validate_eligibility(self, national_id: str, dob_str: str, criminal_record: str): + """Validate all eligibility rules. Raises ValueError on failure.""" + for c in self.store.candidates.values(): + if c.national_id == national_id: + raise ValueError("A candidate with this National ID already exists.") + + age = calculate_age(dob_str) + if age < MIN_CANDIDATE_AGE: + raise ValueError(f"Candidate must be at least {MIN_CANDIDATE_AGE} years old. Current age: {age}") + if age > MAX_CANDIDATE_AGE: + raise ValueError(f"Candidate must not be older than {MAX_CANDIDATE_AGE}. Current age: {age}") + if criminal_record.lower() == "yes": + raise ValueError("Candidates with criminal records are not eligible.") + return age + + def create(self, data: dict, created_by: str) -> Candidate: + """Create and store a new candidate. Returns the created Candidate.""" + age = self.validate_eligibility(data["national_id"], data["date_of_birth"], data["criminal_record"]) + + candidate = Candidate( + id=self.store.candidate_id_counter, + full_name=data["full_name"], + national_id=data["national_id"], + date_of_birth=data["date_of_birth"], + age=age, + gender=data["gender"], + education=data["education"], + party=data["party"], + manifesto=data["manifesto"], + address=data["address"], + phone=data["phone"], + email=data["email"], + has_criminal_record=False, + years_experience=data.get("years_experience", 0), + is_active=True, + is_approved=True, + created_at=str(datetime.datetime.now()), + created_by=created_by, + ) + self.store.candidates[candidate.id] = candidate + self.store.candidate_id_counter += 1 + log_action(self.store, "CREATE_CANDIDATE", created_by, f"Created candidate: {candidate.full_name} (ID: {candidate.id})") + self.store.save() + return candidate + + def update(self, candidate_id: int, updates: dict, updated_by: str) -> Candidate: + """Apply non-empty updates to a candidate. Returns updated Candidate.""" + candidate = self.store.candidates.get(candidate_id) + if not candidate: + raise ValueError("Candidate not found.") + + for field, value in updates.items(): + if value: + setattr(candidate, field, value) + + log_action(self.store, "UPDATE_CANDIDATE", updated_by, f"Updated candidate: {candidate.full_name} (ID: {candidate_id})") + self.store.save() + return candidate + + def deactivate(self, candidate_id: int, deleted_by: str): + """Soft-delete a candidate by deactivating them.""" + candidate = self.store.candidates.get(candidate_id) + if not candidate: + raise ValueError("Candidate not found.") + + for poll in self.store.polls.values(): + if poll.status == "open": + for pos in poll.positions: + if candidate_id in pos.candidate_ids: + raise ValueError(f"Cannot delete — candidate is in active poll: {poll.title}") + + candidate.is_active = False + log_action(self.store, "DELETE_CANDIDATE", deleted_by, f"Deactivated candidate: {candidate.full_name} (ID: {candidate_id})") + self.store.save() + + def search(self, field: str, term) -> list: + """Search candidates by field. Returns a list of matching Candidates.""" + results = [] + for c in self.store.candidates.values(): + if field == "name" and term.lower() in c.full_name.lower(): + results.append(c) + elif field == "party" and term.lower() in c.party.lower(): + results.append(c) + elif field == "education" and c.education == term: + results.append(c) + elif field == "age_range": + min_age, max_age = term + if min_age <= c.age <= max_age: + results.append(c) + return results diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/poll_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/poll_service.py new file mode 100644 index 0000000..f5ba740 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/poll_service.py @@ -0,0 +1,179 @@ +""" +Poll and Position service — business logic for creating, updating, +opening/closing polls and managing positions. No UI output. +""" + +import datetime + +from config import MIN_CANDIDATE_AGE +from models.poll import Poll, PollPosition, Position +from services.audit_service import log_action + + +class PositionService: + def __init__(self, store): + self.store = store + + def create(self, data: dict, created_by: str) -> Position: + if not data["title"]: + raise ValueError("Title cannot be empty.") + if data["level"].lower() not in ["national", "regional", "local"]: + raise ValueError("Invalid level. Choose: National, Regional, or Local.") + if data["max_winners"] <= 0: + raise ValueError("Must have at least 1 winner seat.") + + position = Position( + id=self.store.position_id_counter, + title=data["title"], + description=data.get("description", ""), + level=data["level"].capitalize(), + max_winners=data["max_winners"], + min_candidate_age=data.get("min_candidate_age", MIN_CANDIDATE_AGE), + is_active=True, + created_at=str(datetime.datetime.now()), + created_by=created_by, + ) + self.store.positions[position.id] = position + self.store.position_id_counter += 1 + log_action(self.store, "CREATE_POSITION", created_by, f"Created position: {position.title} (ID: {position.id})") + self.store.save() + return position + + def update(self, position_id: int, updates: dict, updated_by: str) -> Position: + position = self.store.positions.get(position_id) + if not position: + raise ValueError("Position not found.") + for field, value in updates.items(): + if value: + setattr(position, field, value) + log_action(self.store, "UPDATE_POSITION", updated_by, f"Updated position: {position.title}") + self.store.save() + return position + + def deactivate(self, position_id: int, deactivated_by: str): + position = self.store.positions.get(position_id) + if not position: + raise ValueError("Position not found.") + for poll in self.store.polls.values(): + for pp in poll.positions: + if pp.position_id == position_id and poll.status == "open": + raise ValueError(f"Cannot delete — in active poll: {poll.title}") + position.is_active = False + log_action(self.store, "DELETE_POSITION", deactivated_by, f"Deactivated position: {position.title}") + self.store.save() + + +class PollService: + def __init__(self, store): + self.store = store + + def create(self, data: dict, created_by: str) -> Poll: + if not data["title"]: + raise ValueError("Title cannot be empty.") + try: + sd = datetime.datetime.strptime(data["start_date"], "%Y-%m-%d") + ed = datetime.datetime.strptime(data["end_date"], "%Y-%m-%d") + except ValueError: + raise ValueError("Invalid date format. Use YYYY-MM-DD.") + if ed <= sd: + raise ValueError("End date must be after start date.") + + poll_positions = [] + for pid in data["position_ids"]: + pos = self.store.positions.get(pid) + if not pos or not pos.is_active: + continue + poll_positions.append(PollPosition( + position_id=pid, + position_title=pos.title, + candidate_ids=[], + max_winners=pos.max_winners, + )) + if not poll_positions: + raise ValueError("No valid positions selected.") + + poll = Poll( + id=self.store.poll_id_counter, + title=data["title"], + description=data.get("description", ""), + election_type=data["election_type"], + start_date=data["start_date"], + end_date=data["end_date"], + positions=poll_positions, + station_ids=data["station_ids"], + status="draft", + total_votes_cast=0, + created_at=str(datetime.datetime.now()), + created_by=created_by, + ) + self.store.polls[poll.id] = poll + self.store.poll_id_counter += 1 + log_action(self.store, "CREATE_POLL", created_by, f"Created poll: {poll.title} (ID: {poll.id})") + self.store.save() + return poll + + def update(self, poll_id: int, updates: dict, updated_by: str) -> Poll: + poll = self.store.polls.get(poll_id) + if not poll: + raise ValueError("Poll not found.") + if poll.status == "open": + raise ValueError("Cannot update an open poll. Close it first.") + if poll.status == "closed" and poll.total_votes_cast > 0: + raise ValueError("Cannot update a poll with recorded votes.") + for field, value in updates.items(): + if value: + setattr(poll, field, value) + log_action(self.store, "UPDATE_POLL", updated_by, f"Updated poll: {poll.title}") + self.store.save() + return poll + + def delete(self, poll_id: int, deleted_by: str): + poll = self.store.polls.get(poll_id) + if not poll: + raise ValueError("Poll not found.") + if poll.status == "open": + raise ValueError("Cannot delete an open poll. Close it first.") + title = poll.title + del self.store.polls[poll_id] + self.store.votes = [v for v in self.store.votes if v.poll_id != poll_id] + log_action(self.store, "DELETE_POLL", deleted_by, f"Deleted poll: {title}") + self.store.save() + + def open(self, poll_id: int, opened_by: str) -> Poll: + poll = self.store.polls.get(poll_id) + if not poll: + raise ValueError("Poll not found.") + if not any(pos.candidate_ids for pos in poll.positions): + raise ValueError("Cannot open — no candidates assigned.") + poll.status = "open" + log_action(self.store, "OPEN_POLL", opened_by, f"Opened poll: {poll.title}") + self.store.save() + return poll + + def close(self, poll_id: int, closed_by: str) -> Poll: + poll = self.store.polls.get(poll_id) + if not poll: + raise ValueError("Poll not found.") + poll.status = "closed" + log_action(self.store, "CLOSE_POLL", closed_by, f"Closed poll: {poll.title}") + self.store.save() + return poll + + def reopen(self, poll_id: int, reopened_by: str) -> Poll: + poll = self.store.polls.get(poll_id) + if not poll: + raise ValueError("Poll not found.") + poll.status = "open" + log_action(self.store, "REOPEN_POLL", reopened_by, f"Reopened poll: {poll.title}") + self.store.save() + return poll + + def assign_candidates(self, poll_id: int, position_index: int, candidate_ids: list, assigned_by: str): + poll = self.store.polls.get(poll_id) + if not poll: + raise ValueError("Poll not found.") + if poll.status == "open": + raise ValueError("Cannot modify candidates of an open poll.") + poll.positions[position_index].candidate_ids = candidate_ids + log_action(self.store, "ASSIGN_CANDIDATES", assigned_by, f"Updated candidates for poll: {poll.title}") + self.store.save() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/results_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/results_service.py new file mode 100644 index 0000000..0d55160 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/results_service.py @@ -0,0 +1,125 @@ +""" +Results service — computes vote tallies, turnout statistics, +station-wise breakdowns, and demographic summaries. No UI output. +""" + + +class ResultsService: + def __init__(self, store): + self.store = store + + def tally_position(self, poll_id: int, position_id: int) -> dict: + """ + Count votes for a single position in a poll. + Returns { candidate_id: count, '_abstain': count, '_total': count }. + """ + tally = {} + abstain_count = 0 + total = 0 + + for vote in self.store.votes: + if vote.poll_id == poll_id and vote.position_id == position_id: + total += 1 + if vote.abstained: + abstain_count += 1 + else: + tally[vote.candidate_id] = tally.get(vote.candidate_id, 0) + 1 + + return {"counts": tally, "abstain": abstain_count, "total": total} + + def poll_turnout(self, poll_id: int) -> dict: + """Return turnout info: eligible voter count, votes cast, percentage.""" + poll = self.store.polls.get(poll_id) + if not poll: + return {"eligible": 0, "voted": 0, "percentage": 0.0} + + eligible = sum( + 1 for v in self.store.voters.values() + if v.is_verified and v.is_active and v.station_id in poll.station_ids + ) + voted = poll.total_votes_cast + percentage = (voted / eligible * 100) if eligible > 0 else 0.0 + return {"eligible": eligible, "voted": voted, "percentage": percentage} + + def station_turnout(self, poll_id: int, station_id: int) -> dict: + """Return turnout for a specific station within a poll.""" + station_votes = [ + v for v in self.store.votes + if v.poll_id == poll_id and v.station_id == station_id + ] + voters_who_voted = len(set(v.voter_id for v in station_votes)) + registered_at_station = sum( + 1 for v in self.store.voters.values() + if v.station_id == station_id and v.is_verified and v.is_active + ) + percentage = (voters_who_voted / registered_at_station * 100) if registered_at_station > 0 else 0.0 + return { + "registered": registered_at_station, + "voted": voters_who_voted, + "percentage": percentage, + "votes": station_votes, + } + + def system_overview(self) -> dict: + """Return high-level counts for the statistics dashboard.""" + s = self.store + return { + "total_candidates": len(s.candidates), + "active_candidates": sum(1 for c in s.candidates.values() if c.is_active), + "total_voters": len(s.voters), + "verified_voters": sum(1 for v in s.voters.values() if v.is_verified), + "active_voters": sum(1 for v in s.voters.values() if v.is_active), + "total_stations": len(s.voting_stations), + "active_stations": sum(1 for st in s.voting_stations.values() if st.is_active), + "total_polls": len(s.polls), + "open_polls": sum(1 for p in s.polls.values() if p.status == "open"), + "closed_polls": sum(1 for p in s.polls.values() if p.status == "closed"), + "draft_polls": sum(1 for p in s.polls.values() if p.status == "draft"), + "total_votes": len(s.votes), + } + + def voter_demographics(self) -> dict: + """Return gender counts and age group distribution.""" + gender_counts = {} + age_groups = {"18-25": 0, "26-35": 0, "36-45": 0, "46-55": 0, "56-65": 0, "65+": 0} + + for v in self.store.voters.values(): + g = v.gender or "?" + gender_counts[g] = gender_counts.get(g, 0) + 1 + age = v.age or 0 + if age <= 25: age_groups["18-25"] += 1 + elif age <= 35: age_groups["26-35"] += 1 + elif age <= 45: age_groups["36-45"] += 1 + elif age <= 55: age_groups["46-55"] += 1 + elif age <= 65: age_groups["56-65"] += 1 + else: age_groups["65+"] += 1 + + return {"gender": gender_counts, "age_groups": age_groups} + + def station_load(self) -> list: + """Return load info for every station.""" + result = [] + for sid, station in self.store.voting_stations.items(): + vc = sum(1 for v in self.store.voters.values() if v.station_id == sid) + load_pct = (vc / station.capacity * 100) if station.capacity > 0 else 0 + result.append({ + "station": station, + "voter_count": vc, + "load_pct": load_pct, + "overloaded": load_pct > 100, + }) + return result + + def party_distribution(self) -> dict: + counts = {} + for c in self.store.candidates.values(): + if c.is_active: + counts[c.party] = counts.get(c.party, 0) + 1 + return counts + + def education_distribution(self) -> dict: + counts = {} + for c in self.store.candidates.values(): + if c.is_active: + counts[c.education] = counts.get(c.education, 0) + 1 + return counts diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/station_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/station_service.py new file mode 100644 index 0000000..8a30e4b --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/station_service.py @@ -0,0 +1,65 @@ +""" +Station service — business logic for voting station CRUD. +No UI output. +""" + +import datetime + +from models.station import VotingStation +from services.audit_service import log_action + + +class StationService: + def __init__(self, store): + self.store = store + + def create(self, data: dict, created_by: str) -> VotingStation: + if not data["name"]: + raise ValueError("Station name cannot be empty.") + if not data["location"]: + raise ValueError("Location cannot be empty.") + if data["capacity"] <= 0: + raise ValueError("Capacity must be positive.") + + station = VotingStation( + id=self.store.station_id_counter, + name=data["name"], + location=data["location"], + region=data["region"], + capacity=data["capacity"], + registered_voters=0, + supervisor=data.get("supervisor", ""), + contact=data.get("contact", ""), + opening_time=data.get("opening_time", ""), + closing_time=data.get("closing_time", ""), + is_active=True, + created_at=str(datetime.datetime.now()), + created_by=created_by, + ) + self.store.voting_stations[station.id] = station + self.store.station_id_counter += 1 + log_action(self.store, "CREATE_STATION", created_by, f"Created station: {station.name} (ID: {station.id})") + self.store.save() + return station + + def update(self, station_id: int, updates: dict, updated_by: str) -> VotingStation: + station = self.store.voting_stations.get(station_id) + if not station: + raise ValueError("Station not found.") + for field, value in updates.items(): + if value: + setattr(station, field, value) + log_action(self.store, "UPDATE_STATION", updated_by, f"Updated station: {station.name} (ID: {station_id})") + self.store.save() + return station + + def deactivate(self, station_id: int, deactivated_by: str): + station = self.store.voting_stations.get(station_id) + if not station: + raise ValueError("Station not found.") + station.is_active = False + log_action(self.store, "DELETE_STATION", deactivated_by, f"Deactivated station: {station.name}") + self.store.save() + + def voter_count(self, station_id: int) -> int: + return sum(1 for v in self.store.voters.values() if v.station_id == station_id) diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/vote_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/vote_service.py new file mode 100644 index 0000000..9724e3c --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/vote_service.py @@ -0,0 +1,74 @@ +""" +Vote service — handles ballot casting with duplicate prevention. +No UI output. +""" + +import datetime +import hashlib + +from models.vote import Vote +from services.audit_service import log_action + + +class VoteService: + def __init__(self, store): + self.store = store + + def get_available_polls(self, voter) -> dict: + """Return polls the voter can still vote in.""" + available = {} + for pid, poll in self.store.polls.items(): + if ( + poll.status == "open" + and pid not in voter.has_voted_in + and voter.station_id in poll.station_ids + ): + available[pid] = poll + return available + + def cast_ballot(self, voter, poll_id: int, selections: list): + """ + Record a voter's ballot selections. + + selections: list of dicts with keys: + position_id, position_title, candidate_id (or None), abstained + """ + poll = self.store.polls.get(poll_id) + if not poll or poll.status != "open": + raise ValueError("This poll is not open for voting.") + if poll_id in voter.has_voted_in: + raise ValueError("You have already voted in this poll.") + + timestamp = str(datetime.datetime.now()) + vote_hash = hashlib.sha256( + f"{voter.id}{poll_id}{timestamp}".encode() + ).hexdigest()[:16] + + for sel in selections: + self.store.votes.append(Vote( + vote_id=vote_hash + str(sel["position_id"]), + poll_id=poll_id, + position_id=sel["position_id"], + candidate_id=sel.get("candidate_id"), + voter_id=voter.id, + station_id=voter.station_id, + timestamp=timestamp, + abstained=sel.get("abstained", False), + )) + + voter.has_voted_in.append(poll_id) + # Sync the change back to the store + self.store.voters[voter.id].has_voted_in.append(poll_id) + poll.total_votes_cast += 1 + + log_action( + self.store, + "CAST_VOTE", + voter.voter_card_number, + f"Voted in poll: {poll.title} (Hash: {vote_hash})", + ) + self.store.save() + return vote_hash + + def get_voter_votes(self, voter_id: int, poll_id: int) -> list: + return [v for v in self.store.votes if v.voter_id == voter_id and v.poll_id == poll_id] diff --git a/evoting_refactored (1)/evoting_refactored/evoting/services/voter_service.py b/evoting_refactored (1)/evoting_refactored/evoting/services/voter_service.py new file mode 100644 index 0000000..d6dbd1f --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/services/voter_service.py @@ -0,0 +1,139 @@ +""" +Voter service — business logic for voter registration, verification, +deactivation, and search. No UI output. +""" + +import datetime +import random +import string + +from config import MIN_VOTER_AGE, MIN_PASSWORD_LENGTH +from models.voter import Voter +from services.auth_service import hash_password +from services.audit_service import log_action + + +def generate_voter_card_number() -> str: + return "".join(random.choices(string.ascii_uppercase + string.digits, k=12)) + + +class VoterService: + def __init__(self, store): + self.store = store + + def register(self, data: dict) -> Voter: + """Validate and register a new voter. Returns created Voter.""" + if not data["full_name"]: + raise ValueError("Name cannot be empty.") + if not data["national_id"]: + raise ValueError("National ID cannot be empty.") + + for v in self.store.voters.values(): + if v.national_id == data["national_id"]: + raise ValueError("A voter with this National ID already exists.") + + try: + dob = datetime.datetime.strptime(data["date_of_birth"], "%Y-%m-%d") + except ValueError: + raise ValueError("Invalid date format. Use YYYY-MM-DD.") + + age = (datetime.datetime.now() - dob).days // 365 + if age < MIN_VOTER_AGE: + raise ValueError(f"You must be at least {MIN_VOTER_AGE} years old to register.") + + if data["gender"] not in ["M", "F", "OTHER"]: + raise ValueError("Invalid gender selection.") + + if len(data["password"]) < MIN_PASSWORD_LENGTH: + raise ValueError(f"Password must be at least {MIN_PASSWORD_LENGTH} characters.") + + if data["password"] != data["confirm_password"]: + raise ValueError("Passwords do not match.") + + station_id = data["station_id"] + station = self.store.voting_stations.get(station_id) + if not station or not station.is_active: + raise ValueError("Invalid station selection.") + + voter_card = generate_voter_card_number() + voter = Voter( + id=self.store.voter_id_counter, + full_name=data["full_name"], + national_id=data["national_id"], + date_of_birth=data["date_of_birth"], + age=age, + gender=data["gender"], + address=data.get("address", ""), + phone=data.get("phone", ""), + email=data.get("email", ""), + password=hash_password(data["password"]), + voter_card_number=voter_card, + station_id=station_id, + is_verified=False, + is_active=True, + has_voted_in=[], + registered_at=str(datetime.datetime.now()), + role="voter", + ) + self.store.voters[voter.id] = voter + self.store.voter_id_counter += 1 + log_action(self.store, "REGISTER", voter.full_name, f"New voter registered with card: {voter_card}") + self.store.save() + return voter + + def verify(self, voter_id: int, verified_by: str): + voter = self.store.voters.get(voter_id) + if not voter: + raise ValueError("Voter not found.") + if voter.is_verified: + raise ValueError("Voter is already verified.") + voter.is_verified = True + log_action(self.store, "VERIFY_VOTER", verified_by, f"Verified voter: {voter.full_name}") + self.store.save() + + def verify_all_pending(self, verified_by: str) -> int: + count = 0 + for voter in self.store.voters.values(): + if not voter.is_verified: + voter.is_verified = True + count += 1 + log_action(self.store, "VERIFY_ALL_VOTERS", verified_by, f"Verified {count} voters") + self.store.save() + return count + + def deactivate(self, voter_id: int, deactivated_by: str): + voter = self.store.voters.get(voter_id) + if not voter: + raise ValueError("Voter not found.") + if not voter.is_active: + raise ValueError("Voter is already deactivated.") + voter.is_active = False + log_action(self.store, "DEACTIVATE_VOTER", deactivated_by, f"Deactivated voter: {voter.full_name}") + self.store.save() + + def change_password(self, voter_id: int, old_password: str, new_password: str, confirm_password: str): + voter = self.store.voters.get(voter_id) + if not voter: + raise ValueError("Voter not found.") + if hash_password(old_password) != voter.password: + raise ValueError("Incorrect current password.") + if len(new_password) < MIN_PASSWORD_LENGTH: + raise ValueError(f"Password must be at least {MIN_PASSWORD_LENGTH} characters.") + if new_password != confirm_password: + raise ValueError("Passwords do not match.") + voter.password = hash_password(new_password) + log_action(self.store, "CHANGE_PASSWORD", voter.voter_card_number, "Password changed") + self.store.save() + + def search(self, field: str, term) -> list: + results = [] + for v in self.store.voters.values(): + if field == "name" and term.lower() in v.full_name.lower(): + results.append(v) + elif field == "card" and term == v.voter_card_number: + results.append(v) + elif field == "national_id" and term == v.national_id: + results.append(v) + elif field == "station" and v.station_id == term: + results.append(v) + return results diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/__init__.py b/evoting_refactored (1)/evoting_refactored/evoting/ui/__init__.py new file mode 100644 index 0000000..67e55b5 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/ui/__init__.py @@ -0,0 +1 @@ +# ui package diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/__init__.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..acee355 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/__init__.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/colors.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/colors.cpython-313.pyc new file mode 100644 index 0000000..26a092e Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/colors.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/components.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/components.cpython-313.pyc new file mode 100644 index 0000000..7e70e0c Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/components.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/prompts.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/prompts.cpython-313.pyc new file mode 100644 index 0000000..91f47d2 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/ui/__pycache__/prompts.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/colors.py b/evoting_refactored (1)/evoting_refactored/evoting/ui/colors.py new file mode 100644 index 0000000..6e00d1d --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/ui/colors.py @@ -0,0 +1,61 @@ +""" +ANSI color codes and theme constants for terminal output. +""" + +import os +import sys + +if sys.platform == "win32": + os.system("") + +# Base styles +RESET = "\033[0m" +BOLD = "\033[1m" +DIM = "\033[2m" +ITALIC = "\033[3m" +UNDERLINE = "\033[4m" + +# Foreground colors +BLACK = "\033[30m" +RED = "\033[31m" +GREEN = "\033[32m" +YELLOW = "\033[33m" +BLUE = "\033[34m" +MAGENTA = "\033[35m" +CYAN = "\033[36m" +WHITE = "\033[37m" +GRAY = "\033[90m" +BRIGHT_RED = "\033[91m" +BRIGHT_GREEN = "\033[92m" +BRIGHT_YELLOW = "\033[93m" +BRIGHT_BLUE = "\033[94m" +BRIGHT_MAGENTA = "\033[95m" +BRIGHT_CYAN = "\033[96m" +BRIGHT_WHITE = "\033[97m" + +# Background colors +BG_RED = "\033[41m" +BG_GREEN = "\033[42m" +BG_YELLOW = "\033[43m" +BG_BLUE = "\033[44m" +BG_MAGENTA = "\033[45m" +BG_CYAN = "\033[46m" +BG_WHITE = "\033[47m" +BG_GRAY = "\033[100m" + +# Theme colors per context +THEME_LOGIN = BRIGHT_CYAN +THEME_ADMIN = BRIGHT_GREEN +THEME_ADMIN_ACCENT = YELLOW +THEME_VOTER = BRIGHT_BLUE +THEME_VOTER_ACCENT = MAGENTA + + +def colored(text, color): + """Wrap text in a color code.""" + return f"{color}{text}{RESET}" + + +def status_badge(text, is_good): + """Return a green or red colored status label.""" + return f"{GREEN}{text}{RESET}" if is_good else f"{RED}{text}{RESET}" diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/components.py b/evoting_refactored (1)/evoting_refactored/evoting/ui/components.py new file mode 100644 index 0000000..140e1f2 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/ui/components.py @@ -0,0 +1,65 @@ +""" +Reusable UI display components: headers, tables, menus, messages. +""" + +from ui.colors import ( + RESET, BOLD, DIM, GRAY, + RED, GREEN, YELLOW, + THEME_ADMIN, +) + + +def header(title, theme_color): + width = 58 + top = f" {theme_color}{'═' * width}{RESET}" + mid = f" {theme_color}{BOLD} {title.center(width - 2)} {RESET}{theme_color} {RESET}" + bot = f" {theme_color}{'═' * width}{RESET}" + print(top) + print(mid) + print(bot) + + +def subheader(title, theme_color): + print(f"\n {theme_color}{BOLD}▸ {title}{RESET}") + + +def table_header(format_str, theme_color): + print(f" {theme_color}{BOLD}{format_str}{RESET}") + + +def table_divider(width, theme_color): + print(f" {theme_color}{'─' * width}{RESET}") + + +def menu_item(number, text, color): + print(f" {color}{BOLD}{number:>3}.{RESET} {text}") + + +def error(msg): + print(f" {RED}{BOLD} {msg}{RESET}") + + +def success(msg): + print(f" {GREEN}{BOLD} {msg}{RESET}") + + +def warning(msg): + print(f" {YELLOW}{BOLD} {msg}{RESET}") + + +def info(msg): + print(f" {GRAY}{msg}{RESET}") +def status_badge(text, status="info"): + """ + Displays a small label showing the status of something + like ACTIVE, CLOSED, VERIFIED, etc. + """ + + if status == "success": + return f"[✓ {text}]" + elif status == "error": + return f"[✗ {text}]" + elif status == "warning": + return f"[! {text}]" + else: + return f"[{text}]" \ No newline at end of file diff --git a/evoting_refactored (1)/evoting_refactored/evoting/ui/prompts.py b/evoting_refactored (1)/evoting_refactored/evoting/ui/prompts.py new file mode 100644 index 0000000..432b803 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/ui/prompts.py @@ -0,0 +1,72 @@ +""" +User input helpers: prompt, masked password input, pause, clear screen. +""" + +import os +import sys + +from ui.colors import RESET, DIM, BRIGHT_WHITE, YELLOW + + +def clear_screen(): + os.system("cls" if os.name == "nt" else "clear") + + +def pause(): + input(f"\n {DIM}Press Enter to continue...{RESET}") + + +def prompt(text): + return input(f" {BRIGHT_WHITE}{text}{RESET}").strip() + + +def masked_input(prompt_text="Password: "): + """Read a password from stdin, masking each character with '*'.""" + print(f" {BRIGHT_WHITE}{prompt_text}{RESET}", end="", flush=True) + password = "" + + if sys.platform == "win32": + import msvcrt + while True: + ch = msvcrt.getwch() + if ch in ("\r", "\n"): + print() + break + elif ch in ("\x08", "\b"): + if password: + password = password[:-1] + sys.stdout.write("\b \b") + sys.stdout.flush() + elif ch == "\x03": + raise KeyboardInterrupt + else: + password += ch + sys.stdout.write(f"{YELLOW}*{RESET}") + sys.stdout.flush() + else: + import tty + import termios + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + try: + tty.setraw(fd) + while True: + ch = sys.stdin.read(1) + if ch in ("\r", "\n"): + print() + break + elif ch in ("\x7f", "\x08"): + if password: + password = password[:-1] + sys.stdout.write("\b \b") + sys.stdout.flush() + elif ch == "\x03": + raise KeyboardInterrupt + else: + password += ch + sys.stdout.write(f"{YELLOW}*{RESET}") + sys.stdout.flush() + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + + return password diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/__init__.py b/evoting_refactored (1)/evoting_refactored/evoting/views/__init__.py new file mode 100644 index 0000000..18ad452 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/views/__init__.py @@ -0,0 +1 @@ +# views package diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/__init__.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..f7276ac Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/__init__.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/admin_view.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/admin_view.cpython-313.pyc new file mode 100644 index 0000000..ec0b3ef Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/admin_view.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/auth_view.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/auth_view.cpython-313.pyc new file mode 100644 index 0000000..28ce222 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/auth_view.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/voter_view.cpython-313.pyc b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/voter_view.cpython-313.pyc new file mode 100644 index 0000000..45d1fc9 Binary files /dev/null and b/evoting_refactored (1)/evoting_refactored/evoting/views/__pycache__/voter_view.cpython-313.pyc differ diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/admin_view.py b/evoting_refactored (1)/evoting_refactored/evoting/views/admin_view.py new file mode 100644 index 0000000..c26b3c4 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/views/admin_view.py @@ -0,0 +1,1311 @@ +""" +Admin view — all admin-facing menus and screens. +Handles only user input and output. Delegates all logic to services. +""" + +from config import REQUIRED_EDUCATION_LEVELS, ADMIN_ROLES, ADMIN_ROLE_LABELS, POSITION_LEVELS +from ui.colors import ( + RESET, BOLD, DIM, ITALIC, GRAY, + GREEN, RED, YELLOW, BLACK, + BRIGHT_YELLOW, BRIGHT_GREEN, BRIGHT_WHITE, + BG_GREEN, THEME_ADMIN, THEME_ADMIN_ACCENT, +) +from ui.components import ( + header, subheader, table_header, table_divider, + menu_item, error, success, warning, info, status_badge, +) +from ui.prompts import clear_screen, pause, prompt, masked_input + + +class AdminView: + def __init__(self, store, session, services): + self.store = store + self.session = session + self.candidate_svc = services["candidate"] + self.station_svc = services["station"] + self.position_svc = services["position"] + self.poll_svc = services["poll"] + self.voter_svc = services["voter"] + self.admin_svc = services["admin"] + self.results_svc = services["results"] + + # ────────────────────────────────────────────── + # Dashboard + # ────────────────────────────────────────────── + def run(self): + while True: + clear_screen() + header("ADMIN DASHBOARD", THEME_ADMIN) + print(f" {THEME_ADMIN} ● {RESET}{BOLD}{self.session.current_user.full_name}{RESET}" + f" {DIM}│ Role: {self.session.current_user.role}{RESET}") + + subheader("Candidate Management", THEME_ADMIN_ACCENT) + menu_item(1, "Create Candidate", THEME_ADMIN) + menu_item(2, "View All Candidates", THEME_ADMIN) + menu_item(3, "Update Candidate", THEME_ADMIN) + menu_item(4, "Delete Candidate", THEME_ADMIN) + menu_item(5, "Search Candidates", THEME_ADMIN) + + subheader("Voting Station Management", THEME_ADMIN_ACCENT) + menu_item(6, "Create Voting Station", THEME_ADMIN) + menu_item(7, "View All Stations", THEME_ADMIN) + menu_item(8, "Update Station", THEME_ADMIN) + menu_item(9, "Delete Station", THEME_ADMIN) + + subheader("Polls & Positions", THEME_ADMIN_ACCENT) + menu_item(10, "Create Position", THEME_ADMIN) + menu_item(11, "View Positions", THEME_ADMIN) + menu_item(12, "Update Position", THEME_ADMIN) + menu_item(13, "Delete Position", THEME_ADMIN) + menu_item(14, "Create Poll", THEME_ADMIN) + menu_item(15, "View All Polls", THEME_ADMIN) + menu_item(16, "Update Poll", THEME_ADMIN) + menu_item(17, "Delete Poll", THEME_ADMIN) + menu_item(18, "Open/Close Poll", THEME_ADMIN) + menu_item(19, "Assign Candidates to Poll", THEME_ADMIN) + + subheader("Voter Management", THEME_ADMIN_ACCENT) + menu_item(20, "View All Voters", THEME_ADMIN) + menu_item(21, "Verify Voter", THEME_ADMIN) + menu_item(22, "Deactivate Voter", THEME_ADMIN) + menu_item(23, "Search Voters", THEME_ADMIN) + + subheader("Admin Management", THEME_ADMIN_ACCENT) + menu_item(24, "Create Admin Account", THEME_ADMIN) + menu_item(25, "View Admins", THEME_ADMIN) + menu_item(26, "Deactivate Admin", THEME_ADMIN) + + subheader("Results & Reports", THEME_ADMIN_ACCENT) + menu_item(27, "View Poll Results", THEME_ADMIN) + menu_item(28, "View Detailed Statistics", THEME_ADMIN) + menu_item(29, "View Audit Log", THEME_ADMIN) + menu_item(30, "Station-wise Results", THEME_ADMIN) + + subheader("System", THEME_ADMIN_ACCENT) + menu_item(31, "Save Data", THEME_ADMIN) + menu_item(32, "Logout", THEME_ADMIN) + print() + choice = prompt("Enter choice: ") + + actions = { + "1": self.create_candidate, "2": self.view_all_candidates, + "3": self.update_candidate, "4": self.delete_candidate, + "5": self.search_candidates, "6": self.create_station, + "7": self.view_all_stations, "8": self.update_station, + "9": self.delete_station, "10": self.create_position, + "11": self.view_positions, "12": self.update_position, + "13": self.delete_position, "14": self.create_poll, + "15": self.view_all_polls, "16": self.update_poll, + "17": self.delete_poll, "18": self.open_close_poll, + "19": self.assign_candidates_to_poll, "20": self.view_all_voters, + "21": self.verify_voter, "22": self.deactivate_voter, + "23": self.search_voters, "24": self.create_admin, + "25": self.view_admins, "26": self.deactivate_admin, + "27": self.view_poll_results, "28": self.view_detailed_statistics, + "29": self.view_audit_log, "30": self.station_wise_results, + } + + if choice in actions: + actions[choice]() + elif choice == "31": + self.store.save() + pause() + elif choice == "32": + from services.audit_service import log_action + log_action(self.store, "LOGOUT", self.session.current_user.username, "Admin logged out") + self.store.save() + break + else: + error("Invalid choice.") + pause() + + # ────────────────────────────────────────────── + # Candidate screens + # ────────────────────────────────────────────── + def create_candidate(self): + clear_screen() + header("CREATE NEW CANDIDATE", THEME_ADMIN) + print() + full_name = prompt("Full Name: ") + if not full_name: + error("Name cannot be empty.") + pause() + return + national_id = prompt("National ID: ") + if not national_id: + error("National ID cannot be empty.") + pause() + return + dob_str = prompt("Date of Birth (YYYY-MM-DD): ") + gender = prompt("Gender (M/F/Other): ").upper() + + subheader("Education Levels", THEME_ADMIN_ACCENT) + for i, level in enumerate(REQUIRED_EDUCATION_LEVELS, 1): + print(f" {THEME_ADMIN}{i}.{RESET} {level}") + try: + edu_choice = int(prompt("Select education level: ")) + if edu_choice < 1 or edu_choice > len(REQUIRED_EDUCATION_LEVELS): + error("Invalid choice.") + pause() + return + education = REQUIRED_EDUCATION_LEVELS[edu_choice - 1] + except ValueError: + error("Invalid input.") + pause() + return + + party = prompt("Political Party/Affiliation: ") + manifesto = prompt("Brief Manifesto/Bio: ") + address = prompt("Address: ") + phone = prompt("Phone: ") + email = prompt("Email: ") + criminal_record = prompt("Has Criminal Record? (yes/no): ") + years_exp_str = prompt("Years of Public Service/Political Experience: ") + try: + years_experience = int(years_exp_str) + except ValueError: + years_experience = 0 + + try: + candidate = self.candidate_svc.create( + data={ + "full_name": full_name, "national_id": national_id, + "date_of_birth": dob_str, "gender": gender, + "education": education, "party": party, + "manifesto": manifesto, "address": address, + "phone": phone, "email": email, + "criminal_record": criminal_record, + "years_experience": years_experience, + }, + created_by=self.session.current_user.username, + ) + print() + success(f"Candidate '{candidate.full_name}' created successfully! ID: {candidate.id}") + except ValueError as e: + error(str(e)) + pause() + + def view_all_candidates(self): + clear_screen() + header("ALL CANDIDATES", THEME_ADMIN) + if not self.store.candidates: + print() + info("No candidates found.") + pause() + return + print() + table_header(f"{'ID':<5} {'Name':<25} {'Party':<20} {'Age':<5} {'Education':<20} {'Status':<10}", THEME_ADMIN) + table_divider(85, THEME_ADMIN) + for c in self.store.candidates.values(): + badge = status_badge("Active", True) if c.is_active else status_badge("Inactive", False) + print(f" {c.id:<5} {c.full_name:<25} {c.party:<20} {c.age:<5} {c.education:<20} {badge}") + print(f"\n {DIM}Total Candidates: {len(self.store.candidates)}{RESET}") + pause() + + def update_candidate(self): + clear_screen() + header("UPDATE CANDIDATE", THEME_ADMIN) + if not self.store.candidates: + print() + info("No candidates found.") + pause() + return + print() + for c in self.store.candidates.values(): + print(f" {THEME_ADMIN}{c.id}.{RESET} {c.full_name} {DIM}({c.party}){RESET}") + try: + cid = int(prompt("\nEnter Candidate ID to update: ")) + except ValueError: + error("Invalid input.") + pause() + return + c = self.store.candidates.get(cid) + if not c: + error("Candidate not found.") + pause() + return + print(f"\n {BOLD}Updating: {c.full_name}{RESET}") + info("Press Enter to keep current value\n") + updates = { + "full_name": prompt(f"Full Name [{c.full_name}]: "), + "party": prompt(f"Party [{c.party}]: "), + "manifesto": prompt(f"Manifesto [{c.manifesto[:50]}...]: "), + "phone": prompt(f"Phone [{c.phone}]: "), + "email": prompt(f"Email [{c.email}]: "), + "address": prompt(f"Address [{c.address}]: "), + } + exp_input = prompt(f"Years Experience [{c.years_experience}]: ") + if exp_input: + try: + updates["years_experience"] = int(exp_input) + except ValueError: + warning("Invalid number, keeping old value.") + try: + self.candidate_svc.update(cid, updates, self.session.current_user.username) + print() + success(f"Candidate '{self.store.candidates[cid].full_name}' updated successfully!") + except ValueError as e: + error(str(e)) + pause() + + def delete_candidate(self): + clear_screen() + header("DELETE CANDIDATE", THEME_ADMIN) + if not self.store.candidates: + print() + info("No candidates found.") + pause() + return + print() + for c in self.store.candidates.values(): + badge = status_badge("Active", True) if c.is_active else status_badge("Inactive", False) + print(f" {THEME_ADMIN}{c.id}.{RESET} {c.full_name} {DIM}({c.party}){RESET} {badge}") + try: + cid = int(prompt("\nEnter Candidate ID to delete: ")) + except ValueError: + error("Invalid input.") + pause() + return + candidate = self.store.candidates.get(cid) + if not candidate: + error("Candidate not found.") + pause() + return + if prompt(f"Are you sure you want to delete '{candidate.full_name}'? (yes/no): ").lower() == "yes": + try: + self.candidate_svc.deactivate(cid, self.session.current_user.username) + print() + success(f"Candidate '{candidate.full_name}' has been deactivated.") + except ValueError as e: + error(str(e)) + else: + info("Deletion cancelled.") + pause() + + def search_candidates(self): + clear_screen() + header("SEARCH CANDIDATES", THEME_ADMIN) + subheader("Search by", THEME_ADMIN_ACCENT) + menu_item(1, "Name", THEME_ADMIN) + menu_item(2, "Party", THEME_ADMIN) + menu_item(3, "Education Level", THEME_ADMIN) + menu_item(4, "Age Range", THEME_ADMIN) + choice = prompt("\nChoice: ") + + try: + if choice == "1": + term = prompt("Enter name to search: ") + results = self.candidate_svc.search("name", term) + elif choice == "2": + term = prompt("Enter party name: ") + results = self.candidate_svc.search("party", term) + elif choice == "3": + subheader("Education Levels", THEME_ADMIN_ACCENT) + for i, level in enumerate(REQUIRED_EDUCATION_LEVELS, 1): + print(f" {THEME_ADMIN}{i}.{RESET} {level}") + edu_choice = int(prompt("Select: ")) + edu = REQUIRED_EDUCATION_LEVELS[edu_choice - 1] + results = self.candidate_svc.search("education", edu) + elif choice == "4": + min_age = int(prompt("Min age: ")) + max_age = int(prompt("Max age: ")) + results = self.candidate_svc.search("age_range", (min_age, max_age)) + else: + error("Invalid choice.") + pause() + return + except (ValueError, IndexError): + error("Invalid input.") + pause() + return + + if not results: + print() + info("No candidates found matching your criteria.") + else: + print(f"\n {BOLD}Found {len(results)} candidate(s):{RESET}") + table_header(f"{'ID':<5} {'Name':<25} {'Party':<20} {'Age':<5} {'Education':<20}", THEME_ADMIN) + table_divider(75, THEME_ADMIN) + for c in results: + print(f" {c.id:<5} {c.full_name:<25} {c.party:<20} {c.age:<5} {c.education:<20}") + pause() + + # ────────────────────────────────────────────── + # Station screens + # ────────────────────────────────────────────── + def create_station(self): + clear_screen() + header("CREATE VOTING STATION", THEME_ADMIN) + print() + name = prompt("Station Name: ") + location = prompt("Location/Address: ") + region = prompt("Region/District: ") + try: + capacity = int(prompt("Voter Capacity: ")) + except ValueError: + error("Invalid capacity.") + pause() + return + supervisor = prompt("Station Supervisor Name: ") + contact = prompt("Contact Phone: ") + opening_time = prompt("Opening Time (e.g. 08:00): ") + closing_time = prompt("Closing Time (e.g. 17:00): ") + try: + station = self.station_svc.create( + data={ + "name": name, "location": location, "region": region, + "capacity": capacity, "supervisor": supervisor, + "contact": contact, "opening_time": opening_time, + "closing_time": closing_time, + }, + created_by=self.session.current_user.username, + ) + print() + success(f"Voting Station '{station.name}' created! ID: {station.id}") + except ValueError as e: + error(str(e)) + pause() + + def view_all_stations(self): + clear_screen() + header("ALL VOTING STATIONS", THEME_ADMIN) + if not self.store.voting_stations: + print() + info("No voting stations found.") + pause() + return + print() + table_header( + f"{'ID':<5} {'Name':<25} {'Location':<25} {'Region':<15} {'Cap.':<8} {'Reg.':<8} {'Status':<10}", + THEME_ADMIN, + ) + table_divider(96, THEME_ADMIN) + for s in self.store.voting_stations.values(): + reg_count = self.station_svc.voter_count(s.id) + badge = status_badge("Active", True) if s.is_active else status_badge("Inactive", False) + print(f" {s.id:<5} {s.name:<25} {s.location:<25} {s.region:<15} {s.capacity:<8} {reg_count:<8} {badge}") + print(f"\n {DIM}Total Stations: {len(self.store.voting_stations)}{RESET}") + pause() + + def update_station(self): + clear_screen() + header("UPDATE VOTING STATION", THEME_ADMIN) + if not self.store.voting_stations: + print() + info("No stations found.") + pause() + return + print() + for s in self.store.voting_stations.values(): + print(f" {THEME_ADMIN}{s.id}.{RESET} {s.name} {DIM}- {s.location}{RESET}") + try: + sid = int(prompt("\nEnter Station ID to update: ")) + except ValueError: + error("Invalid input.") + pause() + return + s = self.store.voting_stations.get(sid) + if not s: + error("Station not found.") + pause() + return + print(f"\n {BOLD}Updating: {s.name}{RESET}") + info("Press Enter to keep current value\n") + cap_input = prompt(f"Capacity [{s.capacity}]: ") + capacity = None + if cap_input: + try: + capacity = int(cap_input) + except ValueError: + warning("Invalid number, keeping old value.") + updates = { + "name": prompt(f"Name [{s.name}]: "), + "location": prompt(f"Location [{s.location}]: "), + "region": prompt(f"Region [{s.region}]: "), + "supervisor": prompt(f"Supervisor [{s.supervisor}]: "), + "contact": prompt(f"Contact [{s.contact}]: "), + } + if capacity: + updates["capacity"] = capacity + try: + self.station_svc.update(sid, updates, self.session.current_user.username) + print() + success(f"Station '{self.store.voting_stations[sid].name}' updated successfully!") + except ValueError as e: + error(str(e)) + pause() + + def delete_station(self): + clear_screen() + header("DELETE VOTING STATION", THEME_ADMIN) + if not self.store.voting_stations: + print() + info("No stations found.") + pause() + return + print() + for s in self.store.voting_stations.values(): + badge = status_badge("Active", True) if s.is_active else status_badge("Inactive", False) + print(f" {THEME_ADMIN}{s.id}.{RESET} {s.name} {DIM}({s.location}){RESET} {badge}") + try: + sid = int(prompt("\nEnter Station ID to delete: ")) + except ValueError: + error("Invalid input.") + pause() + return + if sid not in self.store.voting_stations: + error("Station not found.") + pause() + return + voter_count = self.station_svc.voter_count(sid) + if voter_count > 0: + warning(f"{voter_count} voters are registered at this station.") + if prompt("Proceed with deactivation? (yes/no): ").lower() != "yes": + info("Cancelled.") + pause() + return + if prompt(f"Confirm deactivation of '{self.store.voting_stations[sid].name}'? (yes/no): ").lower() == "yes": + try: + self.station_svc.deactivate(sid, self.session.current_user.username) + print() + success(f"Station '{self.store.voting_stations[sid].name}' deactivated.") + except ValueError as e: + error(str(e)) + else: + info("Cancelled.") + pause() + + # ────────────────────────────────────────────── + # Position screens + # ────────────────────────────────────────────── + def create_position(self): + clear_screen() + header("CREATE POSITION", THEME_ADMIN) + print() + title = prompt("Position Title (e.g. President, Governor, Senator): ") + if not title: + error("Title cannot be empty.") + pause() + return + description = prompt("Description: ") + level = prompt("Level (National/Regional/Local): ") + try: + max_winners = int(prompt("Number of winners/seats: ")) + except ValueError: + error("Invalid number.") + pause() + return + min_age_str = prompt(f"Minimum candidate age [25]: ") + min_age = int(min_age_str) if min_age_str.isdigit() else 25 + try: + position = self.position_svc.create( + data={ + "title": title, "description": description, + "level": level, "max_winners": max_winners, + "min_candidate_age": min_age, + }, + created_by=self.session.current_user.username, + ) + print() + success(f"Position '{position.title}' created! ID: {position.id}") + except ValueError as e: + error(str(e)) + pause() + + def view_positions(self): + clear_screen() + header("ALL POSITIONS", THEME_ADMIN) + if not self.store.positions: + print() + info("No positions found.") + pause() + return + print() + table_header( + f"{'ID':<5} {'Title':<25} {'Level':<12} {'Seats':<8} {'Min Age':<10} {'Status':<10}", + THEME_ADMIN, + ) + table_divider(70, THEME_ADMIN) + for p in self.store.positions.values(): + badge = status_badge("Active", True) if p.is_active else status_badge("Inactive", False) + print(f" {p.id:<5} {p.title:<25} {p.level:<12} {p.max_winners:<8} {p.min_candidate_age:<10} {badge}") + print(f"\n {DIM}Total Positions: {len(self.store.positions)}{RESET}") + pause() + + def update_position(self): + clear_screen() + header("UPDATE POSITION", THEME_ADMIN) + if not self.store.positions: + print() + info("No positions found.") + pause() + return + print() + for p in self.store.positions.values(): + print(f" {THEME_ADMIN}{p.id}.{RESET} {p.title} {DIM}({p.level}){RESET}") + try: + pid = int(prompt("\nEnter Position ID to update: ")) + except ValueError: + error("Invalid input.") + pause() + return + p = self.store.positions.get(pid) + if not p: + error("Position not found.") + pause() + return + print(f"\n {BOLD}Updating: {p.title}{RESET}") + info("Press Enter to keep current value\n") + updates = { + "title": prompt(f"Title [{p.title}]: "), + "description": prompt(f"Description [{p.description[:50]}]: "), + } + new_level = prompt(f"Level [{p.level}]: ") + if new_level and new_level.lower() in POSITION_LEVELS: + updates["level"] = new_level.capitalize() + seats_str = prompt(f"Seats [{p.max_winners}]: ") + if seats_str: + try: + updates["max_winners"] = int(seats_str) + except ValueError: + warning("Keeping old value.") + try: + self.position_svc.update(pid, updates, self.session.current_user.username) + print() + success("Position updated!") + except ValueError as e: + error(str(e)) + pause() + + def delete_position(self): + clear_screen() + header("DELETE POSITION", THEME_ADMIN) + if not self.store.positions: + print() + info("No positions found.") + pause() + return + print() + for p in self.store.positions.values(): + print(f" {THEME_ADMIN}{p.id}.{RESET} {p.title} {DIM}({p.level}){RESET}") + try: + pid = int(prompt("\nEnter Position ID to delete: ")) + except ValueError: + error("Invalid input.") + pause() + return + if pid not in self.store.positions: + error("Position not found.") + pause() + return + if prompt(f"Confirm deactivation of '{self.store.positions[pid].title}'? (yes/no): ").lower() == "yes": + try: + self.position_svc.deactivate(pid, self.session.current_user.username) + print() + success("Position deactivated.") + except ValueError as e: + error(str(e)) + pause() + + # ────────────────────────────────────────────── + # Poll screens + # ────────────────────────────────────────────── + def create_poll(self): + clear_screen() + header("CREATE POLL / ELECTION", THEME_ADMIN) + print() + title = prompt("Poll/Election Title: ") + if not title: + error("Title cannot be empty.") + pause() + return + description = prompt("Description: ") + election_type = prompt("Election Type (General/Primary/By-election/Referendum): ") + start_date = prompt("Start Date (YYYY-MM-DD): ") + end_date = prompt("End Date (YYYY-MM-DD): ") + + active_positions = {pid: p for pid, p in self.store.positions.items() if p.is_active} + if not active_positions: + error("No active positions. Create positions first.") + pause() + return + + subheader("Available Positions", THEME_ADMIN_ACCENT) + for p in active_positions.values(): + print(f" {THEME_ADMIN}{p.id}.{RESET} {p.title} {DIM}({p.level}) - {p.max_winners} seat(s){RESET}") + try: + selected_ids = [int(x.strip()) for x in prompt("\nEnter Position IDs (comma-separated): ").split(",")] + except ValueError: + error("Invalid input.") + pause() + return + + active_stations = {sid: s for sid, s in self.store.voting_stations.items() if s.is_active} + if not active_stations: + error("No active voting stations.") + pause() + return + + subheader("Available Voting Stations", THEME_ADMIN_ACCENT) + for s in active_stations.values(): + print(f" {THEME_ADMIN}{s.id}.{RESET} {s.name} {DIM}({s.location}){RESET}") + + if prompt("\nUse all active stations? (yes/no): ").lower() == "yes": + selected_station_ids = list(active_stations.keys()) + else: + try: + selected_station_ids = [int(x.strip()) for x in prompt("Enter Station IDs (comma-separated): ").split(",")] + except ValueError: + error("Invalid input.") + pause() + return + + try: + poll = self.poll_svc.create( + data={ + "title": title, "description": description, + "election_type": election_type, "start_date": start_date, + "end_date": end_date, "position_ids": selected_ids, + "station_ids": selected_station_ids, + }, + created_by=self.session.current_user.username, + ) + print() + success(f"Poll '{poll.title}' created! ID: {poll.id}") + warning("Status: DRAFT — Assign candidates and then open the poll.") + except ValueError as e: + error(str(e)) + pause() + + def view_all_polls(self): + clear_screen() + header("ALL POLLS / ELECTIONS", THEME_ADMIN) + if not self.store.polls: + print() + info("No polls found.") + pause() + return + for poll in self.store.polls.values(): + sc = GREEN if poll.status == "open" else (YELLOW if poll.status == "draft" else RED) + print(f"\n {BOLD}{THEME_ADMIN}Poll #{poll.id}: {poll.title}{RESET}") + print(f" {DIM}Type:{RESET} {poll.election_type} {DIM}│ Status:{RESET} {sc}{BOLD}{poll.status.upper()}{RESET}") + print(f" {DIM}Period:{RESET} {poll.start_date} to {poll.end_date} {DIM}│ Votes:{RESET} {poll.total_votes_cast}") + for pos in poll.positions: + cand_names = [ + self.store.candidates[cid].full_name + for cid in pos.candidate_ids + if cid in self.store.candidates + ] + cand_display = ", ".join(cand_names) if cand_names else f"{DIM}None assigned{RESET}" + print(f" {THEME_ADMIN_ACCENT}▸{RESET} {pos.position_title}: {cand_display}") + print(f"\n {DIM}Total Polls: {len(self.store.polls)}{RESET}") + pause() + + def update_poll(self): + clear_screen() + header("UPDATE POLL", THEME_ADMIN) + if not self.store.polls: + print() + info("No polls found.") + pause() + return + print() + for poll in self.store.polls.values(): + sc = GREEN if poll.status == "open" else (YELLOW if poll.status == "draft" else RED) + print(f" {THEME_ADMIN}{poll.id}.{RESET} {poll.title} {sc}({poll.status}){RESET}") + try: + pid = int(prompt("\nEnter Poll ID to update: ")) + except ValueError: + error("Invalid input.") + pause() + return + poll = self.store.polls.get(pid) + if not poll: + error("Poll not found.") + pause() + return + print(f"\n {BOLD}Updating: {poll.title}{RESET}") + info("Press Enter to keep current value\n") + updates = { + "title": prompt(f"Title [{poll.title}]: "), + "description": prompt(f"Description [{poll.description[:50]}]: "), + "election_type": prompt(f"Election Type [{poll.election_type}]: "), + "start_date": prompt(f"Start Date [{poll.start_date}]: "), + "end_date": prompt(f"End Date [{poll.end_date}]: "), + } + try: + self.poll_svc.update(pid, updates, self.session.current_user.username) + print() + success("Poll updated!") + except ValueError as e: + error(str(e)) + pause() + + def delete_poll(self): + clear_screen() + header("DELETE POLL", THEME_ADMIN) + if not self.store.polls: + print() + info("No polls found.") + pause() + return + print() + for poll in self.store.polls.values(): + print(f" {THEME_ADMIN}{poll.id}.{RESET} {poll.title} {DIM}({poll.status}){RESET}") + try: + pid = int(prompt("\nEnter Poll ID to delete: ")) + except ValueError: + error("Invalid input.") + pause() + return + if pid not in self.store.polls: + error("Poll not found.") + pause() + return + if self.store.polls[pid].total_votes_cast > 0: + warning(f"This poll has {self.store.polls[pid].total_votes_cast} votes recorded.") + if prompt(f"Confirm deletion of '{self.store.polls[pid].title}'? (yes/no): ").lower() == "yes": + try: + title = self.store.polls[pid].title + self.poll_svc.delete(pid, self.session.current_user.username) + print() + success(f"Poll '{title}' deleted.") + except ValueError as e: + error(str(e)) + pause() + + def open_close_poll(self): + clear_screen() + header("OPEN / CLOSE POLL", THEME_ADMIN) + if not self.store.polls: + print() + info("No polls found.") + pause() + return + print() + for poll in self.store.polls.values(): + sc = GREEN if poll.status == "open" else (YELLOW if poll.status == "draft" else RED) + print(f" {THEME_ADMIN}{poll.id}.{RESET} {poll.title} {sc}{BOLD}{poll.status.upper()}{RESET}") + try: + pid = int(prompt("\nEnter Poll ID: ")) + except ValueError: + error("Invalid input.") + pause() + return + poll = self.store.polls.get(pid) + if not poll: + error("Poll not found.") + pause() + return + user = self.session.current_user.username + try: + if poll.status == "draft": + if prompt(f"Open poll '{poll.title}'? Voting will begin. (yes/no): ").lower() == "yes": + self.poll_svc.open(pid, user) + print() + success(f"Poll '{poll.title}' is now OPEN for voting!") + elif poll.status == "open": + if prompt(f"Close poll '{poll.title}'? No more votes accepted. (yes/no): ").lower() == "yes": + self.poll_svc.close(pid, user) + print() + success(f"Poll '{poll.title}' is now CLOSED.") + elif poll.status == "closed": + info("This poll is already closed.") + if prompt("Reopen it? (yes/no): ").lower() == "yes": + self.poll_svc.reopen(pid, user) + print() + success("Poll reopened!") + except ValueError as e: + error(str(e)) + pause() + + def assign_candidates_to_poll(self): + clear_screen() + header("ASSIGN CANDIDATES TO POLL", THEME_ADMIN) + if not self.store.polls or not self.store.candidates: + print() + info("No polls or candidates found.") + pause() + return + print() + for poll in self.store.polls.values(): + print(f" {THEME_ADMIN}{poll.id}.{RESET} {poll.title} {DIM}({poll.status}){RESET}") + try: + pid = int(prompt("\nEnter Poll ID: ")) + except ValueError: + error("Invalid input.") + pause() + return + poll = self.store.polls.get(pid) + if not poll: + error("Poll not found.") + pause() + return + if poll.status == "open": + error("Cannot modify candidates of an open poll.") + pause() + return + + for i, pos in enumerate(poll.positions): + subheader(f"Position: {pos.position_title}", THEME_ADMIN_ACCENT) + current_names = [ + f"{cid}: {self.store.candidates[cid].full_name}" + for cid in pos.candidate_ids + if cid in self.store.candidates + ] + if current_names: + print(f" {DIM}Current:{RESET} {', '.join(current_names)}") + else: + info("No candidates assigned yet.") + + pos_data = self.store.positions.get(pos.position_id, None) + min_age = pos_data.min_candidate_age if pos_data else 25 + eligible = { + cid: c for cid, c in self.store.candidates.items() + if c.is_active and c.is_approved and c.age >= min_age + } + + if not eligible: + info("No eligible candidates found.") + continue + + subheader("Available Candidates", THEME_ADMIN) + for c in eligible.values(): + marker = f" {GREEN}[ASSIGNED]{RESET}" if c.id in pos.candidate_ids else "" + print(f" {THEME_ADMIN}{c.id}.{RESET} {c.full_name} {DIM}({c.party}) - Age: {c.age}, Edu: {c.education}{RESET}{marker}") + + if prompt(f"\nModify candidates for {pos.position_title}? (yes/no): ").lower() == "yes": + try: + new_ids = [int(x.strip()) for x in prompt("Enter Candidate IDs (comma-separated): ").split(",")] + valid_ids = [nid for nid in new_ids if nid in eligible] + for nid in new_ids: + if nid not in eligible: + warning(f"Candidate {nid} not eligible. Skipping.") + self.poll_svc.assign_candidates(pid, i, valid_ids, self.session.current_user.username) + success(f"{len(valid_ids)} candidate(s) assigned.") + except ValueError: + error("Invalid input. Skipping this position.") + pause() + + # ────────────────────────────────────────────── + # Voter management screens + # ────────────────────────────────────────────── + def view_all_voters(self): + clear_screen() + header("ALL REGISTERED VOTERS", THEME_ADMIN) + if not self.store.voters: + print() + info("No voters registered.") + pause() + return + print() + table_header( + f"{'ID':<5} {'Name':<25} {'Card Number':<15} {'Stn':<6} {'Verified':<10} {'Active':<8}", + THEME_ADMIN, + ) + table_divider(70, THEME_ADMIN) + for v in self.store.voters.values(): + verified = status_badge("Yes", True) if v.is_verified else status_badge("No", False) + active = status_badge("Yes", True) if v.is_active else status_badge("No", False) + print(f" {v.id:<5} {v.full_name:<25} {v.voter_card_number:<15} {v.station_id:<6} {verified:<19} {active}") + verified_count = sum(1 for v in self.store.voters.values() if v.is_verified) + unverified_count = len(self.store.voters) - verified_count + print(f"\n {DIM}Total: {len(self.store.voters)} │ Verified: {verified_count} │ Unverified: {unverified_count}{RESET}") + pause() + + def verify_voter(self): + clear_screen() + header("VERIFY VOTER", THEME_ADMIN) + unverified = {vid: v for vid, v in self.store.voters.items() if not v.is_verified} + if not unverified: + print() + info("No unverified voters.") + pause() + return + subheader("Unverified Voters", THEME_ADMIN_ACCENT) + for v in unverified.values(): + print(f" {THEME_ADMIN}{v.id}.{RESET} {v.full_name} {DIM}│ NID: {v.national_id} │ Card: {v.voter_card_number}{RESET}") + print() + menu_item(1, "Verify a single voter", THEME_ADMIN) + menu_item(2, "Verify all pending voters", THEME_ADMIN) + choice = prompt("\nChoice: ") + if choice == "1": + try: + vid = int(prompt("Enter Voter ID: ")) + except ValueError: + error("Invalid input.") + pause() + return + try: + self.voter_svc.verify(vid, self.session.current_user.username) + print() + success(f"Voter '{self.store.voters[vid].full_name}' verified!") + except ValueError as e: + error(str(e)) + elif choice == "2": + count = self.voter_svc.verify_all_pending(self.session.current_user.username) + print() + success(f"{count} voters verified!") + pause() + + def deactivate_voter(self): + clear_screen() + header("DEACTIVATE VOTER", THEME_ADMIN) + if not self.store.voters: + print() + info("No voters found.") + pause() + return + print() + try: + vid = int(prompt("Enter Voter ID to deactivate: ")) + except ValueError: + error("Invalid input.") + pause() + return + voter = self.store.voters.get(vid) + if not voter: + error("Voter not found.") + pause() + return + if prompt(f"Deactivate '{voter.full_name}'? (yes/no): ").lower() == "yes": + try: + self.voter_svc.deactivate(vid, self.session.current_user.username) + print() + success("Voter deactivated.") + except ValueError as e: + error(str(e)) + pause() + + def search_voters(self): + clear_screen() + header("SEARCH VOTERS", THEME_ADMIN) + subheader("Search by", THEME_ADMIN_ACCENT) + menu_item(1, "Name", THEME_ADMIN) + menu_item(2, "Voter Card Number", THEME_ADMIN) + menu_item(3, "National ID", THEME_ADMIN) + menu_item(4, "Station", THEME_ADMIN) + choice = prompt("\nChoice: ") + try: + if choice == "1": + results = self.voter_svc.search("name", prompt("Name: ")) + elif choice == "2": + results = self.voter_svc.search("card", prompt("Card Number: ")) + elif choice == "3": + results = self.voter_svc.search("national_id", prompt("National ID: ")) + elif choice == "4": + sid = int(prompt("Station ID: ")) + results = self.voter_svc.search("station", sid) + else: + error("Invalid choice.") + pause() + return + except ValueError: + error("Invalid input.") + pause() + return + if not results: + print() + info("No voters found.") + else: + print(f"\n {BOLD}Found {len(results)} voter(s):{RESET}") + for v in results: + verified = status_badge("Verified", True) if v.is_verified else status_badge("Unverified", False) + print( + f" {THEME_ADMIN}ID:{RESET} {v.id} {DIM}│{RESET} {v.full_name}" + f" {DIM}│ Card:{RESET} {v.voter_card_number} {DIM}│{RESET} {verified}" + ) + pause() + + # ────────────────────────────────────────────── + # Admin management screens + # ────────────────────────────────────────────── + def create_admin(self): + clear_screen() + header("CREATE ADMIN ACCOUNT", THEME_ADMIN) + if not self.session.is_super_admin(): + print() + error("Only super admins can create admin accounts.") + pause() + return + print() + username = prompt("Username: ") + full_name = prompt("Full Name: ") + email = prompt("Email: ") + password = masked_input("Password: ").strip() + subheader("Available Roles", THEME_ADMIN_ACCENT) + for key, role in ADMIN_ROLES.items(): + label = ADMIN_ROLE_LABELS.get(role, "") + menu_item(int(key), f"{role} {DIM}─ {label}{RESET}", THEME_ADMIN) + role_choice = prompt("\nSelect role (1-4): ") + role = ADMIN_ROLES.get(role_choice) + if not role: + error("Invalid role.") + pause() + return + try: + admin = self.admin_svc.create( + data={"username": username, "full_name": full_name, "email": email, "password": password, "role": role}, + created_by=self.session.current_user.username, + ) + print() + success(f"Admin '{admin.username}' created with role: {admin.role}") + except ValueError as e: + error(str(e)) + pause() + + def view_admins(self): + clear_screen() + header("ALL ADMIN ACCOUNTS", THEME_ADMIN) + print() + table_header(f"{'ID':<5} {'Username':<20} {'Full Name':<25} {'Role':<20} {'Active':<8}", THEME_ADMIN) + table_divider(78, THEME_ADMIN) + for a in self.store.admins.values(): + active = status_badge("Yes", True) if a.is_active else status_badge("No", False) + print(f" {a.id:<5} {a.username:<20} {a.full_name:<25} {a.role:<20} {active}") + print(f"\n {DIM}Total Admins: {len(self.store.admins)}{RESET}") + pause() + + def deactivate_admin(self): + clear_screen() + header("DEACTIVATE ADMIN", THEME_ADMIN) + if not self.session.is_super_admin(): + print() + error("Only super admins can deactivate admins.") + pause() + return + print() + for a in self.store.admins.values(): + badge = status_badge("Active", True) if a.is_active else status_badge("Inactive", False) + print(f" {THEME_ADMIN}{a.id}.{RESET} {a.username} {DIM}({a.role}){RESET} {badge}") + try: + aid = int(prompt("\nEnter Admin ID to deactivate: ")) + except ValueError: + error("Invalid input.") + pause() + return + if prompt(f"Deactivate '{self.store.admins.get(aid, {}).username if aid in self.store.admins else '?'}'? (yes/no): ").lower() == "yes": + try: + self.admin_svc.deactivate(aid, self.session.current_user.id, self.session.current_user.username) + print() + success("Admin deactivated.") + except ValueError as e: + error(str(e)) + pause() + + # ────────────────────────────────────────────── + # Results & reports screens + # ────────────────────────────────────────────── + def view_poll_results(self): + clear_screen() + header("POLL RESULTS", THEME_ADMIN) + if not self.store.polls: + print() + info("No polls found.") + pause() + return + print() + for poll in self.store.polls.values(): + sc = GREEN if poll.status == "open" else (YELLOW if poll.status == "draft" else RED) + print(f" {THEME_ADMIN}{poll.id}.{RESET} {poll.title} {sc}({poll.status}){RESET}") + try: + pid = int(prompt("\nEnter Poll ID: ")) + except ValueError: + error("Invalid input.") + pause() + return + poll = self.store.polls.get(pid) + if not poll: + error("Poll not found.") + pause() + return + + print() + header(f"RESULTS: {poll.title}", THEME_ADMIN) + sc = GREEN if poll.status == "open" else RED + print(f" {DIM}Status:{RESET} {sc}{BOLD}{poll.status.upper()}{RESET} {DIM}│ Votes:{RESET} {BOLD}{poll.total_votes_cast}{RESET}") + + turnout = self.results_svc.poll_turnout(pid) + tc = GREEN if turnout["percentage"] > 50 else (YELLOW if turnout["percentage"] > 25 else RED) + print(f" {DIM}Eligible:{RESET} {turnout['eligible']} {DIM}│ Turnout:{RESET} {tc}{BOLD}{turnout['percentage']:.1f}%{RESET}") + + for pos in poll.positions: + subheader(f"{pos.position_title} (Seats: {pos.max_winners})", THEME_ADMIN_ACCENT) + tally = self.results_svc.tally_position(pid, pos.position_id) + self._render_tally(tally, pos.max_winners, THEME_ADMIN) + pause() + + def _render_tally(self, tally: dict, max_winners: int, bar_color): + vote_counts = tally["counts"] + abstain_count = tally["abstain"] + total = tally["total"] + if not vote_counts: + info(" No votes recorded for this position.") + return + for rank, (cid, count) in enumerate( + sorted(vote_counts.items(), key=lambda x: x[1], reverse=True), 1 + ): + cand = self.store.candidates.get(cid, None) + name = cand.full_name if cand else "?" + party = cand.party if cand else "?" + pct = (count / total * 100) if total > 0 else 0 + bl = int(pct / 2) + bar = f"{bar_color}{'█' * bl}{GRAY}{'░' * (50 - bl)}{RESET}" + winner = f" {BG_GREEN}{BLACK}{BOLD} ★ WINNER {RESET}" if rank <= max_winners else "" + print(f" {BOLD}{rank}. {name}{RESET} {DIM}({party}){RESET}") + print(f" {bar} {BOLD}{count}{RESET} ({pct:.1f}%){winner}") + if abstain_count > 0: + pct = (abstain_count / total * 100) if total > 0 else 0 + print(f" {GRAY}Abstained: {abstain_count} ({pct:.1f}%){RESET}") + + def view_detailed_statistics(self): + clear_screen() + header("DETAILED STATISTICS", THEME_ADMIN) + overview = self.results_svc.system_overview() + + subheader("SYSTEM OVERVIEW", THEME_ADMIN_ACCENT) + print(f" {THEME_ADMIN}Candidates:{RESET} {overview['total_candidates']} {DIM}(Active: {overview['active_candidates']}){RESET}") + print(f" {THEME_ADMIN}Voters:{RESET} {overview['total_voters']} {DIM}(Verified: {overview['verified_voters']}, Active: {overview['active_voters']}){RESET}") + print(f" {THEME_ADMIN}Stations:{RESET} {overview['total_stations']} {DIM}(Active: {overview['active_stations']}){RESET}") + print(f" {THEME_ADMIN}Polls:{RESET} {overview['total_polls']} " + f"{DIM}({GREEN}Open: {overview['open_polls']}{RESET}{DIM}, {RED}Closed: {overview['closed_polls']}{RESET}{DIM}, {YELLOW}Draft: {overview['draft_polls']}{RESET}{DIM}){RESET}") + print(f" {THEME_ADMIN}Total Votes:{RESET} {overview['total_votes']}") + + demo = self.results_svc.voter_demographics() + subheader("VOTER DEMOGRAPHICS", THEME_ADMIN_ACCENT) + tv = overview["total_voters"] + for g, count in demo["gender"].items(): + pct = (count / tv * 100) if tv > 0 else 0 + print(f" {g}: {count} ({pct:.1f}%)") + print(f" {BOLD}Age Distribution:{RESET}") + for group, count in demo["age_groups"].items(): + pct = (count / tv * 100) if tv > 0 else 0 + print(f" {group:>5}: {count:>3} ({pct:>5.1f}%) {THEME_ADMIN}{'█' * int(pct / 2)}{RESET}") + + subheader("STATION LOAD", THEME_ADMIN_ACCENT) + for item in self.results_svc.station_load(): + s = item["station"] + lp = item["load_pct"] + lc = RED if lp > 100 else (YELLOW if lp > 75 else GREEN) + st = f"{RED}{BOLD}OVERLOADED{RESET}" if item["overloaded"] else f"{GREEN}OK{RESET}" + print(f" {s.name}: {item['voter_count']}/{s.capacity} {lc}({lp:.0f}%){RESET} {st}") + + subheader("CANDIDATE PARTY DISTRIBUTION", THEME_ADMIN_ACCENT) + for party, count in sorted(self.results_svc.party_distribution().items(), key=lambda x: x[1], reverse=True): + print(f" {party}: {BOLD}{count}{RESET} candidate(s)") + + subheader("CANDIDATE EDUCATION LEVELS", THEME_ADMIN_ACCENT) + for edu, count in self.results_svc.education_distribution().items(): + print(f" {edu}: {BOLD}{count}{RESET}") + pause() + + def station_wise_results(self): + clear_screen() + header("STATION-WISE RESULTS", THEME_ADMIN) + if not self.store.polls: + print() + info("No polls found.") + pause() + return + print() + for poll in self.store.polls.values(): + sc = GREEN if poll.status == "open" else (YELLOW if poll.status == "draft" else RED) + print(f" {THEME_ADMIN}{poll.id}.{RESET} {poll.title} {sc}({poll.status}){RESET}") + try: + pid = int(prompt("\nEnter Poll ID: ")) + except ValueError: + error("Invalid input.") + pause() + return + poll = self.store.polls.get(pid) + if not poll: + error("Poll not found.") + pause() + return + print() + header(f"STATION RESULTS: {poll.title}", THEME_ADMIN) + for sid in poll.station_ids: + if sid not in self.store.voting_stations: + continue + station = self.store.voting_stations[sid] + subheader(f"{station.name} ({station.location})", BRIGHT_WHITE) + st_data = self.results_svc.station_turnout(pid, sid) + tc = GREEN if st_data["percentage"] > 50 else (YELLOW if st_data["percentage"] > 25 else RED) + print( + f" {DIM}Registered:{RESET} {st_data['registered']} " + f"{DIM}│ Voted:{RESET} {st_data['voted']} " + f"{DIM}│ Turnout:{RESET} {tc}{BOLD}{st_data['percentage']:.1f}%{RESET}" + ) + for pos in poll.positions: + print(f" {THEME_ADMIN_ACCENT}▸ {pos.position_title}:{RESET}") + pos_votes = [v for v in st_data["votes"] if v.position_id == pos.position_id] + vc = {} + ac = 0 + for v in pos_votes: + if v.abstained: + ac += 1 + else: + vc[v.candidate_id] = vc.get(v.candidate_id, 0) + 1 + total = sum(vc.values()) + ac + for cid, count in sorted(vc.items(), key=lambda x: x[1], reverse=True): + cand = self.store.candidates.get(cid) + name = cand.full_name if cand else "?" + party = cand.party if cand else "?" + pct = (count / total * 100) if total > 0 else 0 + print(f" {name} {DIM}({party}){RESET}: {BOLD}{count}{RESET} ({pct:.1f}%)") + if ac > 0: + pct = (ac / total * 100) if total > 0 else 0 + print(f" {GRAY}Abstained: {ac} ({pct:.1f}%){RESET}") + pause() + + def view_audit_log(self): + clear_screen() + header("AUDIT LOG", THEME_ADMIN) + if not self.store.audit_log: + print() + info("No audit records.") + pause() + return + print(f"\n {DIM}Total Records: {len(self.store.audit_log)}{RESET}") + subheader("Filter", THEME_ADMIN_ACCENT) + menu_item(1, "Last 20 entries", THEME_ADMIN) + menu_item(2, "All entries", THEME_ADMIN) + menu_item(3, "Filter by action type", THEME_ADMIN) + menu_item(4, "Filter by user", THEME_ADMIN) + choice = prompt("\nChoice: ") + entries = self.store.audit_log + + if choice == "1": + entries = self.store.audit_log[-20:] + elif choice == "3": + action_types = list(set(e["action"] for e in self.store.audit_log)) + for i, at in enumerate(action_types, 1): + print(f" {THEME_ADMIN}{i}.{RESET} {at}") + try: + at_choice = int(prompt("Select action type: ")) + entries = [e for e in self.store.audit_log if e["action"] == action_types[at_choice - 1]] + except (ValueError, IndexError): + error("Invalid choice.") + pause() + return + elif choice == "4": + uf = prompt("Enter username/card number: ") + entries = [e for e in self.store.audit_log if uf.lower() in e["user"].lower()] + + print() + table_header(f"{'Timestamp':<22} {'Action':<25} {'User':<20} {'Details'}", THEME_ADMIN) + table_divider(100, THEME_ADMIN) + for entry in entries: + ac = ( + GREEN if "CREATE" in entry["action"] or entry["action"] == "LOGIN" + else RED if "DELETE" in entry["action"] or "DEACTIVATE" in entry["action"] + else YELLOW if "UPDATE" in entry["action"] + else RESET + ) + print( + f" {DIM}{entry['timestamp'][:19]}{RESET} " + f"{ac}{entry['action']:<25}{RESET} " + f"{entry['user']:<20} " + f"{DIM}{entry['details'][:50]}{RESET}" + ) + pause() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/auth_view.py b/evoting_refactored (1)/evoting_refactored/evoting/views/auth_view.py new file mode 100644 index 0000000..ec765de --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/views/auth_view.py @@ -0,0 +1,142 @@ +""" +Auth view — login screen and voter self-registration screen. +Handles only user input/output. Delegates logic to services. +""" + +from ui.colors import ( + RESET, BOLD, DIM, BRIGHT_YELLOW, BRIGHT_BLUE, + THEME_LOGIN, THEME_ADMIN, THEME_VOTER, +) +from ui.components import header, menu_item, error, success, warning, info, subheader +from ui.prompts import clear_screen, pause, prompt, masked_input + + +class AuthView: + def __init__(self, store, session, voter_svc): + self.store = store + self.session = session + self.voter_svc = voter_svc + + def show_login_menu(self) -> bool: + """ + Display the main login menu. Returns True when a user successfully + logs in, False otherwise (so the main loop retries). + """ + clear_screen() + header("E-VOTING SYSTEM", THEME_LOGIN) + print() + menu_item(1, "Login as Admin", THEME_LOGIN) + menu_item(2, "Login as Voter", THEME_LOGIN) + menu_item(3, "Register as Voter", THEME_LOGIN) + menu_item(4, "Exit", THEME_LOGIN) + print() + choice = prompt("Enter choice: ") + + if choice == "1": + return self._admin_login() + elif choice == "2": + return self._voter_login() + elif choice == "3": + self._register_voter() + return False + elif choice == "4": + print() + info("Goodbye!") + self.store.save() + exit() + else: + error("Invalid choice.") + pause() + return False + + def _admin_login(self) -> bool: + clear_screen() + header("ADMIN LOGIN", THEME_ADMIN) + print() + username = prompt("Username: ") + password = masked_input("Password: ").strip() + ok, msg, _ = self.session.login_admin(username, password, self.store.admins) + if ok: + from services.audit_service import log_action + log_action(self.store, "LOGIN", username, "Admin login successful") + print() + success(msg) + pause() + return True + else: + from services.audit_service import log_action + log_action(self.store, "LOGIN_FAILED", username, msg) + error(msg) + pause() + return False + + def _voter_login(self) -> bool: + clear_screen() + header("VOTER LOGIN", THEME_VOTER) + print() + voter_card = prompt("Voter Card Number: ") + password = masked_input("Password: ").strip() + ok, msg, _ = self.session.login_voter(voter_card, password, self.store.voters) + if ok: + from services.audit_service import log_action + log_action(self.store, "LOGIN", voter_card, "Voter login successful") + print() + success(msg) + pause() + return True + else: + from services.audit_service import log_action + log_action(self.store, "LOGIN_FAILED", voter_card, msg) + error(msg) + pause() + return False + + def _register_voter(self): + clear_screen() + header("VOTER REGISTRATION", THEME_VOTER) + print() + full_name = prompt("Full Name: ") + national_id = prompt("National ID Number: ") + dob_str = prompt("Date of Birth (YYYY-MM-DD): ") + gender = prompt("Gender (M/F/Other): ").upper() + address = prompt("Residential Address: ") + phone = prompt("Phone Number: ") + email = prompt("Email Address: ") + password = masked_input("Create Password: ").strip() + confirm_password = masked_input("Confirm Password: ").strip() + + if not self.store.voting_stations: + error("No voting stations available. Contact admin.") + pause() + return + + subheader("Available Voting Stations", THEME_VOTER) + for s in self.store.voting_stations.values(): + if s.is_active: + print(f" {BRIGHT_BLUE}{s.id}.{RESET} {s.name} {DIM}- {s.location}{RESET}") + + try: + station_choice = int(prompt("\nSelect your voting station ID: ")) + except ValueError: + error("Invalid input.") + pause() + return + + try: + voter = self.voter_svc.register( + data={ + "full_name": full_name, "national_id": national_id, + "date_of_birth": dob_str, "gender": gender, + "address": address, "phone": phone, "email": email, + "password": password, "confirm_password": confirm_password, + "station_id": station_choice, + } + ) + print() + success("Registration successful!") + print(f" {BOLD}Your Voter Card Number: {BRIGHT_YELLOW}{voter.voter_card_number}{RESET}") + warning("IMPORTANT: Save this number! You need it to login.") + info("Your registration is pending admin verification.") + except ValueError as e: + error(str(e)) + pause() diff --git a/evoting_refactored (1)/evoting_refactored/evoting/views/voter_view.py b/evoting_refactored (1)/evoting_refactored/evoting/views/voter_view.py new file mode 100644 index 0000000..0e80719 --- /dev/null +++ b/evoting_refactored (1)/evoting_refactored/evoting/views/voter_view.py @@ -0,0 +1,294 @@ +""" +Voter view — all voter-facing menus and screens. +Handles only user input and output. Delegates all logic to services. +""" + +from ui.colors import ( + RESET, BOLD, DIM, ITALIC, GRAY, + GREEN, RED, YELLOW, BLACK, + BRIGHT_GREEN, BRIGHT_YELLOW, BRIGHT_CYAN, BRIGHT_BLUE, + BG_GREEN, THEME_VOTER, THEME_VOTER_ACCENT, +) +from ui.components import ( + header, subheader, menu_item, error, success, warning, info, status_badge, +) +from ui.prompts import clear_screen, pause, prompt, masked_input + + +class VoterView: + def __init__(self, store, session, services): + self.store = store + self.session = session + self.vote_svc = services["vote"] + self.voter_svc = services["voter"] + self.results_svc = services["results"] + + def run(self): + while True: + clear_screen() + header("VOTER DASHBOARD", THEME_VOTER) + voter = self.session.current_user + station_name = self.store.voting_stations.get(voter.station_id, None) + station_label = station_name.name if station_name else "Unknown" + print(f" {THEME_VOTER} ● {RESET}{BOLD}{voter.full_name}{RESET}") + print(f" {DIM} Card: {voter.voter_card_number} │ Station: {station_label}{RESET}") + print() + menu_item(1, "View Open Polls", THEME_VOTER) + menu_item(2, "Cast Vote", THEME_VOTER) + menu_item(3, "View My Voting History", THEME_VOTER) + menu_item(4, "View Results (Closed Polls)", THEME_VOTER) + menu_item(5, "View My Profile", THEME_VOTER) + menu_item(6, "Change Password", THEME_VOTER) + menu_item(7, "Logout", THEME_VOTER) + print() + choice = prompt("Enter choice: ") + + if choice == "1": + self.view_open_polls() + elif choice == "2": + self.cast_vote() + elif choice == "3": + self.view_voting_history() + elif choice == "4": + self.view_closed_results() + elif choice == "5": + self.view_profile() + elif choice == "6": + self.change_password() + elif choice == "7": + from services.audit_service import log_action + log_action(self.store, "LOGOUT", voter.voter_card_number, "Voter logged out") + self.store.save() + break + else: + error("Invalid choice.") + pause() + + def view_open_polls(self): + clear_screen() + header("OPEN POLLS", THEME_VOTER) + voter = self.session.current_user + open_polls = {pid: p for pid, p in self.store.polls.items() if p.status == "open"} + if not open_polls: + print() + info("No open polls at this time.") + pause() + return + for pid, poll in open_polls.items(): + already_voted = pid in voter.has_voted_in + vs = f" {GREEN}[VOTED]{RESET}" if already_voted else f" {YELLOW}[NOT YET VOTED]{RESET}" + print(f"\n {BOLD}{THEME_VOTER}Poll #{poll.id}: {poll.title}{RESET}{vs}") + print(f" {DIM}Type:{RESET} {poll.election_type} {DIM}│ Period:{RESET} {poll.start_date} to {poll.end_date}") + for pos in poll.positions: + print(f" {THEME_VOTER_ACCENT}▸{RESET} {BOLD}{pos.position_title}{RESET}") + for cid in pos.candidate_ids: + if cid in self.store.candidates: + c = self.store.candidates[cid] + print(f" {DIM}•{RESET} {c.full_name} {DIM}({c.party}) │ Age: {c.age} │ Edu: {c.education}{RESET}") + pause() + + def cast_vote(self): + clear_screen() + header("CAST YOUR VOTE", THEME_VOTER) + voter = self.session.current_user + available_polls = self.vote_svc.get_available_polls(voter) + if not available_polls: + print() + info("No available polls to vote in.") + pause() + return + + subheader("Available Polls", THEME_VOTER_ACCENT) + for poll in available_polls.values(): + print(f" {THEME_VOTER}{poll.id}.{RESET} {poll.title} {DIM}({poll.election_type}){RESET}") + + try: + pid = int(prompt("\nSelect Poll ID to vote: ")) + except ValueError: + error("Invalid input.") + pause() + return + if pid not in available_polls: + error("Invalid poll selection.") + pause() + return + + poll = self.store.polls[pid] + print() + header(f"Voting: {poll.title}", THEME_VOTER) + info("Please select ONE candidate for each position.\n") + + selections = [] + for pos in poll.positions: + subheader(pos.position_title, THEME_VOTER_ACCENT) + if not pos.candidate_ids: + info("No candidates for this position.") + continue + + for idx, cid in enumerate(pos.candidate_ids, 1): + if cid in self.store.candidates: + c = self.store.candidates[cid] + print(f" {THEME_VOTER}{BOLD}{idx}.{RESET} {c.full_name} {DIM}({c.party}){RESET}") + print(f" {DIM}Age: {c.age} │ Edu: {c.education} │ Exp: {c.years_experience} yrs{RESET}") + if c.manifesto: + print(f" {ITALIC}{DIM}{c.manifesto[:80]}...{RESET}") + print(f" {GRAY}{BOLD}0.{RESET} {GRAY}Abstain / Skip{RESET}") + + try: + vote_choice = int(prompt(f"\nYour choice for {pos.position_title}: ")) + except ValueError: + warning("Invalid input. Skipping.") + vote_choice = 0 + + if vote_choice == 0 or not (1 <= vote_choice <= len(pos.candidate_ids)): + if vote_choice != 0: + warning("Invalid choice. Marking as abstain.") + selections.append({ + "position_id": pos.position_id, + "position_title": pos.position_title, + "candidate_id": None, + "abstained": True, + }) + else: + selected_cid = pos.candidate_ids[vote_choice - 1] + selections.append({ + "position_id": pos.position_id, + "position_title": pos.position_title, + "candidate_id": selected_cid, + "candidate_name": self.store.candidates[selected_cid].full_name, + "abstained": False, + }) + + subheader("VOTE SUMMARY", BRIGHT_CYAN) + for sel in selections: + if sel["abstained"]: + print(f" {sel['position_title']}: {GRAY}ABSTAINED{RESET}") + else: + print(f" {sel['position_title']}: {BRIGHT_GREEN}{BOLD}{sel['candidate_name']}{RESET}") + print() + + if prompt("Confirm your votes? This cannot be undone. (yes/no): ").lower() != "yes": + info("Vote cancelled.") + pause() + return + + try: + vote_hash = self.vote_svc.cast_ballot(voter, pid, selections) + print() + success("Your vote has been recorded successfully!") + print(f" {DIM}Vote Reference:{RESET} {BRIGHT_YELLOW}{vote_hash}{RESET}") + print(f" {BRIGHT_CYAN}Thank you for participating in the democratic process!{RESET}") + except ValueError as e: + error(str(e)) + pause() + + def view_voting_history(self): + clear_screen() + header("MY VOTING HISTORY", THEME_VOTER) + voter = self.session.current_user + voted_polls = voter.has_voted_in + if not voted_polls: + print() + info("You have not voted in any polls yet.") + pause() + return + print(f"\n {DIM}You have voted in {len(voted_polls)} poll(s):{RESET}\n") + for pid in voted_polls: + if pid not in self.store.polls: + continue + poll = self.store.polls[pid] + sc = GREEN if poll.status == "open" else RED + print(f" {BOLD}{THEME_VOTER}Poll #{pid}: {poll.title}{RESET}") + print(f" {DIM}Type:{RESET} {poll.election_type} {DIM}│ Status:{RESET} {sc}{poll.status.upper()}{RESET}") + my_votes = self.vote_svc.get_voter_votes(voter.id, pid) + for vr in my_votes: + pos_title = next( + (pos.position_title for pos in poll.positions if pos.position_id == vr.position_id), + "Unknown", + ) + if vr.abstained: + print(f" {THEME_VOTER_ACCENT}▸{RESET} {pos_title}: {GRAY}ABSTAINED{RESET}") + else: + cand = self.store.candidates.get(vr.candidate_id) + name = cand.full_name if cand else "Unknown" + print(f" {THEME_VOTER_ACCENT}▸{RESET} {pos_title}: {BRIGHT_GREEN}{name}{RESET}") + print() + pause() + + def view_closed_results(self): + clear_screen() + header("ELECTION RESULTS", THEME_VOTER) + closed_polls = {pid: p for pid, p in self.store.polls.items() if p.status == "closed"} + if not closed_polls: + print() + info("No closed polls with results.") + pause() + return + for pid, poll in closed_polls.items(): + print(f"\n {BOLD}{THEME_VOTER}{poll.title}{RESET}") + print(f" {DIM}Type:{RESET} {poll.election_type} {DIM}│ Votes:{RESET} {poll.total_votes_cast}") + for pos in poll.positions: + subheader(pos.position_title, THEME_VOTER_ACCENT) + tally = self.results_svc.tally_position(pid, pos.position_id) + vote_counts = tally["counts"] + abstain_count = tally["abstain"] + total = tally["total"] + for rank, (cid, count) in enumerate( + sorted(vote_counts.items(), key=lambda x: x[1], reverse=True), 1 + ): + cand = self.store.candidates.get(cid) + name = cand.full_name if cand else "?" + party = cand.party if cand else "?" + pct = (count / total * 100) if total > 0 else 0 + bar = f"{THEME_VOTER}{'█' * int(pct / 2)}{GRAY}{'░' * (50 - int(pct / 2))}{RESET}" + winner = f" {BG_GREEN}{BLACK}{BOLD} WINNER {RESET}" if rank <= pos.max_winners else "" + print(f" {BOLD}{rank}. {name}{RESET} {DIM}({party}){RESET}") + print(f" {bar} {BOLD}{count}{RESET} ({pct:.1f}%){winner}") + if abstain_count > 0: + pct = (abstain_count / total * 100) if total > 0 else 0 + print(f" {GRAY}Abstained: {abstain_count} ({pct:.1f}%){RESET}") + pause() + + def view_profile(self): + clear_screen() + header("MY PROFILE", THEME_VOTER) + v = self.session.current_user + station = self.store.voting_stations.get(v.station_id) + station_label = station.name if station else "Unknown" + print() + fields = [ + ("Name", v.full_name), + ("National ID", v.national_id), + ("Voter Card", f"{BRIGHT_YELLOW}{v.voter_card_number}{RESET}"), + ("Date of Birth", v.date_of_birth), + ("Age", v.age), + ("Gender", v.gender), + ("Address", v.address), + ("Phone", v.phone), + ("Email", v.email), + ("Station", station_label), + ("Verified", status_badge("Yes", True) if v.is_verified else status_badge("No", False)), + ("Registered", v.registered_at), + ("Polls Voted", len(v.has_voted_in)), + ] + for label, value in fields: + print(f" {THEME_VOTER}{label + ':':<16}{RESET} {value}") + pause() + + def change_password(self): + clear_screen() + header("CHANGE PASSWORD", THEME_VOTER) + voter = self.session.current_user + print() + old_pass = masked_input("Current Password: ").strip() + new_pass = masked_input("New Password: ").strip() + confirm_pass = masked_input("Confirm New Password: ").strip() + try: + self.voter_svc.change_password(voter.id, old_pass, new_pass, confirm_pass) + # Keep session in sync + self.session.current_user.password = self.store.voters[voter.id].password + print() + success("Password changed successfully!") + except ValueError as e: + error(str(e)) + pause()