Developer.md

# 🛠️ Azure Speech Transcription - Developer Guide

## 📋 Table of Contents

- [System Architecture](#-system-architecture)
- [Development Environment](#-development-environment)
- [Deployment Guide](#-deployment-guide)
- [API Documentation](#-api-documentation)
- [Database Schema](#-database-schema)
- [Security Implementation](#-security-implementation)
- [Monitoring & Maintenance](#-monitoring--maintenance)
- [Contributing Guidelines](#-contributing-guidelines)
- [Advanced Configuration](#-advanced-configuration)
- [Troubleshooting](#-troubleshooting)

---

## 🏗️ System Architecture

### Overview

The Azure Speech Transcription service is built with a modern, secure architecture focusing on user privacy, PDPA compliance, and scalability.

```

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐

│   Frontend UI   │    │   Backend API   │    │ Azure Services  │

│   (Gradio)      │◄──►│   (Python)      │◄──►│ Speech & Blob   │

└─────────────────┘    └─────────────────┘    └─────────────────┘

         │                       │                       │

         │                       │                       │

         ▼                       ▼                       ▼

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐

│   User Session  │    │ SQLite Database │    │  User Storage   │

│   Management    │    │   (Metadata)    │    │   (Isolated)    │

└─────────────────┘    └─────────────────┘    └─────────────────┘

```

### Core Components

#### 1. Frontend Layer (`gradio_app.py`)

- **Technology**: Gradio with custom CSS

- **Purpose**: User interface and session management

- **Features**: Authentication, file upload, real-time status, history management



#### 2. Backend Layer (`app_core.py`)
- **Technology**: Python with threading and async processing
- **Purpose**: Business logic, authentication, and Azure integration
- **Features**: User management, transcription processing, PDPA compliance

#### 3. Data Layer
- **Database**: SQLite with Azure Blob backup
- **Storage**: Azure Blob Storage with user separation
- **Security**: User-isolated folders and encrypted connections

#### 4. External Services
- **Azure Speech Services**: Transcription processing
- **Azure Blob Storage**: File and database storage
- **FFmpeg**: Audio/video conversion

### Data Flow

```

1. User uploads file → 2. Authentication check → 3. File validation

        ↓                       ↓                       ↓

8. Download results ← 7. Store transcript ← 6. Process with Azure

        ↑                       ↑                       ↑

9. Update UI status ← 4. Save to user folder ← 5. Background processing

```

---

## 💻 Development Environment

### Prerequisites

- **Python**: 3.8 or higher
- **Azure Account**: With Speech Services and Blob Storage
- **FFmpeg**: For audio/video processing
- **Git**: For version control

### Environment Setup

#### 1. Clone Repository
```bash

git clone <repository-url>

cd azure-speech-transcription

```

#### 2. Virtual Environment
```bash

# Create virtual environment

python -m venv venv



# Activate (Windows)

venv\Scripts\activate



# Activate (macOS/Linux)

source venv/bin/activate

```

#### 3. Install Dependencies
```bash

pip install -r requirements.txt

```

#### 4. Environment Configuration
```bash

# Copy environment template

cp .env.example .env



# Edit with your Azure credentials

nano .env

```

#### 5. Install FFmpeg

**Windows (Chocolatey):**
```bash

choco install ffmpeg

```

**macOS (Homebrew):**
```bash

brew install ffmpeg

```

**Ubuntu/Debian:**
```bash

sudo apt update

sudo apt install ffmpeg

```

#### 6. Verify Installation
```python

python -c "

import gradio as gr

from azure.storage.blob import BlobServiceClient

import subprocess

print('Gradio:', gr.__version__)

print('FFmpeg:', subprocess.run(['ffmpeg', '-version'], capture_output=True).returncode == 0)

print('Azure Blob:', 'OK')

"

```

### Development Server

```bash

# Start development server

python gradio_app.py



# Server will be available at:

# http://localhost:7860

```

### Development Tools

#### Recommended IDE Setup
- **VS Code**: With Python, Azure, and Git extensions
- **PyCharm**: Professional edition with Azure toolkit
- **Vim/Emacs**: With appropriate Python plugins

#### Useful Extensions
```json

{

  "recommendations": [

    "ms-python.python",

    "ms-vscode.azure-cli",

    "ms-azuretools.azure-cli-tools",

    "ms-python.black-formatter",

    "ms-python.flake8"

  ]

}

```

#### Code Quality Tools
```bash

# Install development tools

pip install black flake8 pytest mypy



# Format code

black .



# Lint code

flake8 .



# Type checking

mypy app_core.py gradio_app.py

```

---

## 🚀 Deployment Guide

### Production Deployment Options

#### Option 1: Traditional Server Deployment

**1. Server Preparation**
```bash

# Update system

sudo apt update && sudo apt upgrade -y



# Install Python and dependencies

sudo apt install python3 python3-pip python3-venv nginx ffmpeg -y



# Create application user

sudo useradd -m -s /bin/bash transcription

sudo su - transcription

```

**2. Application Setup**
```bash

# Clone repository

git clone <repository-url> /home/transcription/app

cd /home/transcription/app



# Setup virtual environment

python3 -m venv venv

source venv/bin/activate

pip install -r requirements.txt



# Configure environment

cp .env.example .env

# Edit .env with production values

```

**3. Systemd Service**
```ini

# /etc/systemd/system/transcription.service

[Unit]

Description=Azure Speech Transcription Service

After=network.target



[Service]

Type=simple

User=transcription

Group=transcription

WorkingDirectory=/home/transcription/app

Environment=PATH=/home/transcription/app/venv/bin

ExecStart=/home/transcription/app/venv/bin/python gradio_app.py

Restart=always

RestartSec=10



[Install]

WantedBy=multi-user.target

```

**4. Nginx Configuration**
```nginx

# /etc/nginx/sites-available/transcription

server {

    listen 80;

    server_name your-domain.com;

    client_max_body_size 500M;



    location / {

        proxy_pass http://127.0.0.1:7860;

        proxy_set_header Host $host;

        proxy_set_header X-Real-IP $remote_addr;

        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        proxy_set_header X-Forwarded-Proto $scheme;

        proxy_read_timeout 300s;

        proxy_connect_timeout 75s;

    }

}

```

**5. SSL Certificate**
```bash

# Install Certbot

sudo apt install certbot python3-certbot-nginx -y



# Get SSL certificate

sudo certbot --nginx -d your-domain.com



# Verify auto-renewal

sudo certbot renew --dry-run

```

**6. Start Services**
```bash

# Enable and start application

sudo systemctl enable transcription

sudo systemctl start transcription



# Enable and restart nginx

sudo systemctl enable nginx

sudo systemctl restart nginx



# Check status

sudo systemctl status transcription

sudo systemctl status nginx

```

#### Option 2: Docker Deployment

**1. Dockerfile**
```dockerfile

FROM python:3.9-slim



# Install system dependencies

RUN apt-get update && apt-get install -y \

    ffmpeg \

    && rm -rf /var/lib/apt/lists/*



# Set working directory

WORKDIR /app



# Copy requirements and install Python dependencies

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt



# Copy application code

COPY . .



# Create necessary directories

RUN mkdir -p uploads database temp



# Expose port

EXPOSE 7860



# Run application

CMD ["python", "gradio_app.py"]

```

**2. Docker Compose**
```yaml

# docker-compose.yml

version: '3.8'



services:

  transcription:

    build: .

    ports:

      - "7860:7860"

    environment:

      - AZURE_SPEECH_KEY=${AZURE_SPEECH_KEY}

      - AZURE_SPEECH_KEY_ENDPOINT=${AZURE_SPEECH_KEY_ENDPOINT}

      - AZURE_REGION=${AZURE_REGION}

      - AZURE_BLOB_CONNECTION=${AZURE_BLOB_CONNECTION}

      - AZURE_CONTAINER=${AZURE_CONTAINER}

      - AZURE_BLOB_SAS_TOKEN=${AZURE_BLOB_SAS_TOKEN}

      - ALLOWED_LANGS=${ALLOWED_LANGS}

    volumes:

      - ./uploads:/app/uploads

      - ./database:/app/database

      - ./temp:/app/temp

    restart: unless-stopped



  nginx:

    image: nginx:alpine

    ports:

      - "80:80"

      - "443:443"

    volumes:

      - ./nginx.conf:/etc/nginx/nginx.conf

      - ./ssl:/etc/ssl/certs

    depends_on:

      - transcription

    restart: unless-stopped

```

**3. Deploy with Docker**
```bash

# Build and start

docker-compose up -d



# View logs

docker-compose logs -f transcription



# Update application

git pull

docker-compose build transcription

docker-compose up -d transcription

```

#### Option 3: Cloud Deployment (Azure Container Instances)

**1. Create Container Registry**
```bash

# Create ACR

az acr create --resource-group myResourceGroup \

  --name myregistry --sku Basic



# Login to ACR

az acr login --name myregistry



# Build and push image

docker build -t myregistry.azurecr.io/transcription:latest .

docker push myregistry.azurecr.io/transcription:latest

```

**2. Deploy Container Instance**
```bash

# Create container instance

az container create \

  --resource-group myResourceGroup \

  --name transcription-app \

  --image myregistry.azurecr.io/transcription:latest \

  --cpu 2 --memory 4 \

  --port 7860 \

  --environment-variables \

    AZURE_SPEECH_KEY=$AZURE_SPEECH_KEY \

    AZURE_SPEECH_KEY_ENDPOINT=$AZURE_SPEECH_KEY_ENDPOINT \

    AZURE_REGION=$AZURE_REGION \

    AZURE_BLOB_CONNECTION="$AZURE_BLOB_CONNECTION" \

    AZURE_CONTAINER=$AZURE_CONTAINER \

    AZURE_BLOB_SAS_TOKEN="$AZURE_BLOB_SAS_TOKEN"

```

---

## 📡 API Documentation

### Core Classes and Methods

#### TranscriptionManager Class

**Purpose**: Main service class handling all transcription operations

```python

class TranscriptionManager:

    def __init__(self)

    

    # User Authentication

    def register_user(email: str, username: str, password: str, 

                     gdpr_consent: bool, data_retention_agreed: bool, 

                     marketing_consent: bool) -> Tuple[bool, str, Optional[str]]

    

    def login_user(login: str, password: str) -> Tuple[bool, str, Optional[User]]

    

    # Transcription Operations  

    def submit_transcription(file_bytes: bytes, original_filename: str,

                           user_id: str, language: str, 

                           settings: Dict) -> str

    

    def get_job_status(job_id: str) -> Optional[TranscriptionJob]

    

    # Data Management

    def get_user_history(user_id: str, limit: int) -> List[TranscriptionJob]

    def get_user_stats(user_id: str) -> Dict

    def export_user_data(user_id: str) -> Dict

    def delete_user_account(user_id: str) -> bool

```

#### DatabaseManager Class

**Purpose**: Handle database operations and Azure blob synchronization

```python

class DatabaseManager:

    def __init__(db_path: str = None)

    

    # User Operations

    def create_user(...) -> Tuple[bool, str, Optional[str]]

    def authenticate_user(login: str, password: str) -> Tuple[bool, str, Optional[User]]

    def get_user_by_id(user_id: str) -> Optional[User]

    

    # Job Operations

    def save_job(job: TranscriptionJob)

    def get_job(job_id: str) -> Optional[TranscriptionJob]

    def get_user_jobs(user_id: str, limit: int) -> List[TranscriptionJob]

    def get_pending_jobs() -> List[TranscriptionJob]

```

#### AuthManager Class

**Purpose**: Authentication utilities and validation

```python

class AuthManager:

    @staticmethod

    def hash_password(password: str) -> str

    def verify_password(password: str, password_hash: str) -> bool

    def validate_email(email: str) -> bool

    def validate_username(username: str) -> bool

    def validate_password(password: str) -> Tuple[bool, str]

```

### Data Models

#### User Model
```python

@dataclass

class User:

    user_id: str

    email: str

    username: str

    password_hash: str

    created_at: str

    last_login: Optional[str] = None

    is_active: bool = True

    gdpr_consent: bool = False

    data_retention_agreed: bool = False

    marketing_consent: bool = False

```

#### TranscriptionJob Model
```python

@dataclass

class TranscriptionJob:

    job_id: str

    user_id: str

    original_filename: str

    audio_url: str

    language: str

    status: str  # pending, processing, completed, failed

    created_at: str

    completed_at: Optional[str] = None

    transcript_text: Optional[str] = None

    transcript_url: Optional[str] = None

    error_message: Optional[str] = None

    azure_trans_id: Optional[str] = None

    settings: Optional[Dict] = None

```

### Configuration Parameters

#### Environment Variables
```python

# Required

AZURE_SPEECH_KEY: str

AZURE_SPEECH_KEY_ENDPOINT: str  

AZURE_REGION: str

AZURE_BLOB_CONNECTION: str

AZURE_CONTAINER: str

AZURE_BLOB_SAS_TOKEN: str



# Optional

ALLOWED_LANGS: str  # JSON string

API_VERSION: str = "v3.2"

PASSWORD_SALT: str = "default_salt"

MAX_FILE_SIZE_MB: int = 500

```

#### Transcription Settings
```python

settings = {

    'audio_format': str,           # wav, mp3, etc.

    'diarization_enabled': bool,   # Speaker identification

    'speakers': int,               # Max speakers (1-10)

    'profanity': str,              # masked, removed, raw

    'punctuation': str,            # automatic, dictated, none

    'timestamps': bool,            # Include timestamps

    'lexical': bool,               # Include lexical forms

    'language_id_enabled': bool,   # Auto language detection

    'candidate_locales': List[str] # Language candidates

}

```

---

## 🗄️ Database Schema

### SQLite Database Structure

#### Users Table
```sql

CREATE TABLE users (

    user_id TEXT PRIMARY KEY,

    email TEXT UNIQUE NOT NULL,

    username TEXT UNIQUE NOT NULL,

    password_hash TEXT NOT NULL,

    created_at TEXT NOT NULL,

    last_login TEXT,

    is_active BOOLEAN DEFAULT 1,

    gdpr_consent BOOLEAN DEFAULT 0,

    data_retention_agreed BOOLEAN DEFAULT 0,

    marketing_consent BOOLEAN DEFAULT 0

);



-- Indexes

CREATE INDEX idx_users_email ON users(email);

CREATE INDEX idx_users_username ON users(username);

```

#### Transcriptions Table
```sql

CREATE TABLE transcriptions (

    job_id TEXT PRIMARY KEY,

    user_id TEXT NOT NULL,

    original_filename TEXT NOT NULL,

    audio_url TEXT,

    language TEXT NOT NULL,

    status TEXT NOT NULL,

    created_at TEXT NOT NULL,

    completed_at TEXT,

    transcript_text TEXT,

    transcript_url TEXT,

    error_message TEXT,

    azure_trans_id TEXT,

    settings TEXT,

    FOREIGN KEY (user_id) REFERENCES users (user_id)

);



-- Indexes

CREATE INDEX idx_transcriptions_user_id ON transcriptions(user_id);

CREATE INDEX idx_transcriptions_status ON transcriptions(status);

CREATE INDEX idx_transcriptions_created_at ON transcriptions(created_at DESC);

CREATE INDEX idx_transcriptions_user_created ON transcriptions(user_id, created_at DESC);

```

### Azure Blob Storage Structure

```

Container: {AZURE_CONTAINER}/

├── shared/

│   └── database/

│       └── transcriptions.db           # Shared database backup

├── users/

│   ├── {user-id-1}/

│   │   ├── audio/                      # Processed audio files

│   │   │   ├── {job-id-1}.wav

│   │   │   └── {job-id-2}.wav

│   │   ├── transcripts/                # Transcript files

│   │   │   ├── {job-id-1}.txt

│   │   │   └── {job-id-2}.txt

│   │   └── originals/                  # Original uploaded files

│   │       ├── {job-id-1}_{filename}.mp4

│   │       └── {job-id-2}_{filename}.wav

│   └── {user-id-2}/

│       ├── audio/

│       ├── transcripts/

│       └── originals/

```

### Database Operations

#### User Management Queries
```sql

-- Create user

INSERT INTO users (user_id, email, username, password_hash, created_at, 

                   gdpr_consent, data_retention_agreed, marketing_consent)

VALUES (?, ?, ?, ?, ?, ?, ?, ?);



-- Authenticate user

SELECT * FROM users 

WHERE (email = ? OR username = ?) AND is_active = 1;



-- Update last login

UPDATE users SET last_login = ? WHERE user_id = ?;



-- Get user stats

SELECT status, COUNT(*) FROM transcriptions 

WHERE user_id = ? GROUP BY status;

```

#### Job Management Queries
```sql

-- Create job

INSERT INTO transcriptions (job_id, user_id, original_filename, language, 

                           status, created_at, settings)

VALUES (?, ?, ?, ?, 'pending', ?, ?);



-- Update job status

UPDATE transcriptions 

SET status = ?, completed_at = ?, transcript_text = ?, transcript_url = ?

WHERE job_id = ?;



-- Get user jobs

SELECT * FROM transcriptions 

WHERE user_id = ? 

ORDER BY created_at DESC LIMIT ?;



-- Get pending jobs for background processor

SELECT * FROM transcriptions 

WHERE status IN ('pending', 'processing');

```

---

## 🔒 Security Implementation

### Authentication Security

#### Password Security
```python

# Password hashing with salt

def hash_password(password: str) -> str:

    salt = os.environ.get("PASSWORD_SALT", "default_salt")

    return hashlib.sha256((password + salt).encode()).hexdigest()



# Password validation

def validate_password(password: str) -> Tuple[bool, str]:

    if len(password) < 8:

        return False, "Password must be at least 8 characters"

    if not re.search(r'[A-Z]', password):

        return False, "Password must contain uppercase letter"

    if not re.search(r'[a-z]', password):

        return False, "Password must contain lowercase letter"

    if not re.search(r'\d', password):

        return False, "Password must contain number"

    return True, "Valid"

```

#### Session Management
```python

# User session state

session_state = {

    'user_id': str,

    'username': str,

    'logged_in_at': datetime,

    'last_activity': datetime

}



# Session validation

def validate_session(session_state: dict) -> bool:

    if not session_state or 'user_id' not in session_state:

        return False

    

    # Check session timeout (if implemented)

    last_activity = session_state.get('last_activity')

    if last_activity:

        timeout = timedelta(hours=24)  # 24-hour sessions

        if datetime.now() - last_activity > timeout:

            return False

    

    return True

```

### Data Security

#### Access Control
```python

# User data access verification

def verify_user_access(job_id: str, user_id: str) -> bool:

    job = get_job(job_id)

    return job and job.user_id == user_id



# File path security

def get_user_blob_path(user_id: str, blob_type: str, filename: str) -> str:

    # Ensure user can only access their own folder

    safe_filename = os.path.basename(filename)  # Prevent path traversal

    return f"users/{user_id}/{blob_type}/{safe_filename}"

```

#### Data Encryption
```python

# Azure Blob Storage encryption (configured at Azure level)

# - Encryption at rest: Enabled by default

# - Encryption in transit: HTTPS enforced

# - Customer-managed keys: Optional enhancement



# Database encryption (for sensitive fields)

from cryptography.fernet import Fernet



def encrypt_sensitive_data(data: str, key: bytes) -> str:

    f = Fernet(key)

    return f.encrypt(data.encode()).decode()



def decrypt_sensitive_data(encrypted_data: str, key: bytes) -> str:

    f = Fernet(key)

    return f.decrypt(encrypted_data.encode()).decode()

```

### Azure Security

#### Blob Storage Security
```python

# SAS token configuration for least privilege

sas_permissions = BlobSasPermissions(

    read=True,

    write=True,

    delete=True,

    list=True

)



# IP restrictions (optional)

sas_ip_range = "192.168.1.0/24"  # Restrict to specific IP range



# Time-limited tokens

sas_expiry = datetime.utcnow() + timedelta(hours=1)

```

#### Speech Service Security
```python

# Secure API calls

headers = {

    "Ocp-Apim-Subscription-Key": AZURE_SPEECH_KEY,

    "Content-Type": "application/json"

}



# Request timeout and retry logic

response = requests.post(

    url, 

    headers=headers, 

    json=body,

    timeout=30,

    verify=True  # Verify SSL certificates

)

```

### Input Validation

#### File Upload Security
```python

def validate_uploaded_file(file_path: str, max_size: int = 500 * 1024 * 1024) -> Tuple[bool, str]:

    try:

        # Check file exists

        if not os.path.exists(file_path):

            return False, "File not found"

        

        # Check file size

        file_size = os.path.getsize(file_path)

        if file_size > max_size:

            return False, f"File too large: {file_size / 1024 / 1024:.1f}MB"

        

        # Check file type by content (not just extension)

        import magic

        mime_type = magic.from_file(file_path, mime=True)

        allowed_types = ['audio/', 'video/']

        if not any(mime_type.startswith(t) for t in allowed_types):

            return False, f"Invalid file type: {mime_type}"

        

        return True, "Valid"

    

    except Exception as e:

        return False, f"Validation error: {str(e)}"

```

#### SQL Injection Prevention
```python

# Use parameterized queries (already implemented)

cursor.execute(

    "SELECT * FROM users WHERE email = ? AND password_hash = ?",

    (email, password_hash)

)



# Input sanitization

def sanitize_input(user_input: str) -> str:

    # Remove dangerous characters

    import html

    sanitized = html.escape(user_input)

    # Limit length

    return sanitized[:1000]

```

---

## 📊 Monitoring & Maintenance

### Application Monitoring

#### Health Checks
```python

def health_check() -> Dict[str, Any]:

    """System health check endpoint"""

    try:

        # Database check

        db_status = check_database_connection()

        

        # Azure services check

        blob_status = check_blob_storage()

        speech_status = check_speech_service()

        

        # FFmpeg check

        ffmpeg_status = check_ffmpeg_installation()

        

        # Disk space check

        disk_status = check_disk_space()

        

        return {

            'status': 'healthy' if all([db_status, blob_status, speech_status, ffmpeg_status]) else 'unhealthy',

            'timestamp': datetime.now().isoformat(),

            'services': {

                'database': db_status,

                'blob_storage': blob_status,

                'speech_service': speech_status,

                'ffmpeg': ffmpeg_status,

                'disk_space': disk_status

            }

        }

    

    except Exception as e:

        return {

            'status': 'error',

            'timestamp': datetime.now().isoformat(),

            'error': str(e)

        }



def check_database_connection() -> bool:

    try:

        with transcription_manager.db.get_connection() as conn:

            conn.execute("SELECT 1").fetchone()

        return True

    except:

        return False



def check_blob_storage() -> bool:

    try:

        client = BlobServiceClient.from_connection_string(AZURE_BLOB_CONNECTION)

        client.list_containers(max_results=1)

        return True

    except:

        return False

```

#### Logging Configuration
```python

import logging

from logging.handlers import RotatingFileHandler



def setup_logging():

    """Configure application logging"""

    

    # Create formatter

    formatter = logging.Formatter(

        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'

    )

    

    # Console handler

    console_handler = logging.StreamHandler()

    console_handler.setFormatter(formatter)

    console_handler.setLevel(logging.INFO)

    

    # File handler with rotation

    file_handler = RotatingFileHandler(

        'logs/transcription.log',

        maxBytes=10*1024*1024,  # 10MB

        backupCount=5

    )

    file_handler.setFormatter(formatter)

    file_handler.setLevel(logging.DEBUG)

    

    # Configure root logger

    logger = logging.getLogger()

    logger.setLevel(logging.DEBUG)

    logger.addHandler(console_handler)

    logger.addHandler(file_handler)

    

    # Separate logger for sensitive operations

    auth_logger = logging.getLogger('auth')

    auth_handler = RotatingFileHandler(

        'logs/auth.log',

        maxBytes=5*1024*1024,  # 5MB

        backupCount=10

    )

    auth_handler.setFormatter(formatter)

    auth_logger.addHandler(auth_handler)

    auth_logger.setLevel(logging.INFO)

```

#### Performance Monitoring
```python

import time

from functools import wraps



def monitor_performance(func):

    """Decorator to monitor function performance"""

    @wraps(func)

    def wrapper(*args, **kwargs):

        start_time = time.time()

        try:

            result = func(*args, **kwargs)

            duration = time.time() - start_time

            logging.info(f"{func.__name__} completed in {duration:.2f}s")

            return result

        except Exception as e:

            duration = time.time() - start_time

            logging.error(f"{func.__name__} failed after {duration:.2f}s: {str(e)}")

            raise

    return wrapper



# Usage

@monitor_performance

def submit_transcription(self, file_bytes, filename, user_id, language, settings):

    # Implementation here

    pass

```

### Database Maintenance

#### Backup Strategy
```python

def backup_database():

    """Backup database to Azure Blob Storage"""

    try:

        # Create timestamped backup

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

        backup_name = f"shared/backups/transcriptions_backup_{timestamp}.db"

        

        # Upload current database

        blob_client = blob_service.get_blob_client(

            container=AZURE_CONTAINER, 

            blob=backup_name

        )

        

        with open(db_path, "rb") as data:

            blob_client.upload_blob(data)

        

        logging.info(f"Database backup created: {backup_name}")

        

        # Clean old backups (keep last 30 days)

        cleanup_old_backups()

        

    except Exception as e:

        logging.error(f"Database backup failed: {str(e)}")



def cleanup_old_backups():

    """Remove backups older than 30 days"""

    try:

        cutoff_date = datetime.now() - timedelta(days=30)

        container_client = blob_service.get_container_client(AZURE_CONTAINER)

        

        for blob in container_client.list_blobs(name_starts_with="shared/backups/"):

            if blob.last_modified < cutoff_date:

                blob_service.delete_blob(AZURE_CONTAINER, blob.name)

                logging.info(f"Deleted old backup: {blob.name}")

                

    except Exception as e:

        logging.error(f"Backup cleanup failed: {str(e)}")

```

#### Database Optimization
```python

def optimize_database():

    """Optimize database performance"""

    try:

        with transcription_manager.db.get_connection() as conn:

            # Analyze tables

            conn.execute("ANALYZE")

            

            # Vacuum database (compact)

            conn.execute("VACUUM")

            

            # Update statistics

            conn.execute("PRAGMA optimize")

            

        logging.info("Database optimization completed")

        

    except Exception as e:

        logging.error(f"Database optimization failed: {str(e)}")



# Schedule optimization (run weekly)

import schedule



schedule.every().week.do(optimize_database)

schedule.every().day.at("02:00").do(backup_database)

```

### Resource Management

#### Cleanup Tasks
```python

def cleanup_temporary_files():

    """Clean up temporary files older than 24 hours"""

    try:

        cutoff_time = time.time() - (24 * 60 * 60)  # 24 hours ago

        temp_dirs = ['uploads', 'temp']

        

        for temp_dir in temp_dirs:

            if os.path.exists(temp_dir):

                for filename in os.listdir(temp_dir):

                    filepath = os.path.join(temp_dir, filename)

                    if os.path.isfile(filepath) and os.path.getmtime(filepath) < cutoff_time:

                        os.remove(filepath)

                        logging.info(f"Cleaned up temporary file: {filepath}")

                        

    except Exception as e:

        logging.error(f"Temporary file cleanup failed: {str(e)}")



def monitor_disk_space():

    """Monitor and alert on disk space"""

    try:

        import shutil

        total, used, free = shutil.disk_usage("/")

        

        # Convert to GB

        free_gb = free // (1024**3)

        total_gb = total // (1024**3)

        usage_percent = (used / total) * 100

        

        if usage_percent > 85:

            logging.warning(f"High disk usage: {usage_percent:.1f}% ({free_gb}GB free)")

            

        if free_gb < 5:

            logging.critical(f"Low disk space: {free_gb}GB remaining")

            

    except Exception as e:

        logging.error(f"Disk space monitoring failed: {str(e)}")

```

### Monitoring Alerts

#### Email Alerts (Optional)
```python

import smtplib

from email.mime.text import MIMEText



def send_alert(subject: str, message: str):

    """Send email alert for critical issues"""

    try:

        smtp_server = os.environ.get("SMTP_SERVER")

        smtp_port = int(os.environ.get("SMTP_PORT", "587"))

        smtp_user = os.environ.get("SMTP_USER")

        smtp_pass = os.environ.get("SMTP_PASS")

        alert_email = os.environ.get("ALERT_EMAIL")

        

        if not all([smtp_server, smtp_user, smtp_pass, alert_email]):

            return  # Email not configured

        

        msg = MIMEText(message)

        msg['Subject'] = f"[Transcription Service] {subject}"

        msg['From'] = smtp_user

        msg['To'] = alert_email

        

        with smtplib.SMTP(smtp_server, smtp_port) as server:

            server.starttls()

            server.login(smtp_user, smtp_pass)

            server.send_message(msg)

            

    except Exception as e:

        logging.error(f"Failed to send alert: {str(e)}")

```

---

## 🤝 Contributing Guidelines

### Development Workflow

#### 1. Setup Development Environment
```bash

# Fork repository

git clone https://github.com/your-username/azure-speech-transcription.git

cd azure-speech-transcription



# Create feature branch

git checkout -b feature/your-feature-name



# Setup environment

python -m venv venv

source venv/bin/activate  # or venv\Scripts\activate on Windows

pip install -r requirements.txt

pip install -r requirements-dev.txt  # Development dependencies

```

#### 2. Code Quality Standards

**Python Style Guide**
- Follow PEP 8 style guidelines
- Use type hints for function parameters and return values
- Maximum line length: 88 characters (Black formatter)
- Use meaningful variable and function names

**Code Formatting**
```bash

# Install development tools

pip install black flake8 mypy pytest



# Format code

black .



# Check style

flake8 .



# Type checking

mypy app_core.py gradio_app.py



# Run tests

pytest tests/

```

**Documentation Standards**
- All functions must have docstrings
- Include type hints
- Document complex logic with inline comments
- Update README.md for new features

```python

def submit_transcription(

    self, 

    file_bytes: bytes, 

    original_filename: str,

    user_id: str,

    language: str,

    settings: Dict[str, Any]

) -> str:

    """

    Submit a new transcription job for processing.

    

    Args:

        file_bytes: Raw bytes of the audio/video file

        original_filename: Original name of the uploaded file

        user_id: ID of the authenticated user

        language: Language code for transcription (e.g., 'en-US')

        settings: Transcription configuration options

        

    Returns:

        str: Unique job ID for tracking transcription progress

        

    Raises:

        ValueError: If user_id is invalid or file is too large

        ConnectionError: If Azure services are unavailable

    """

```

#### 3. Testing Requirements

**Unit Tests**
```python

import pytest

from unittest.mock import Mock, patch

from app_core import TranscriptionManager, AuthManager



class TestAuthManager:

    def test_password_hashing(self):

        password = "TestPassword123"

        hashed = AuthManager.hash_password(password)

        

        assert hashed != password

        assert AuthManager.verify_password(password, hashed)

        assert not AuthManager.verify_password("wrong", hashed)

    

    def test_email_validation(self):

        assert AuthManager.validate_email("test@example.com")

        assert not AuthManager.validate_email("invalid-email")

        assert not AuthManager.validate_email("")



class TestTranscriptionManager:

    @patch('app_core.BlobServiceClient')

    def test_submit_transcription(self, mock_blob):

        manager = TranscriptionManager()

        

        job_id = manager.submit_transcription(

            b"fake audio data",

            "test.wav",

            "user123",

            "en-US",

            {"audio_format": "wav"}

        )

        

        assert isinstance(job_id, str)

        assert len(job_id) == 36  # UUID length

```

**Integration Tests**
```python

class TestIntegration:

    def test_full_transcription_workflow(self):

        # Test complete workflow from upload to download

        pass

    

    def test_user_registration_and_login(self):

        # Test complete auth workflow

        pass

```

#### 4. Commit Guidelines

**Commit Message Format**
```

type(scope): brief description



Detailed explanation of changes if needed



- List specific changes

- Include any breaking changes

- Reference issue numbers



Closes #123

```

**Commit Types**
- `feat`: New feature
- `fix`: Bug fix
- `docs`: Documentation changes
- `style`: Code style changes (formatting, etc.)
- `refactor`: Code refactoring
- `test`: Adding or updating tests
- `chore`: Maintenance tasks

**Example Commits**
```bash

git commit -m "feat(auth): add password strength validation



- Implement password complexity requirements

- Add client-side validation feedback

- Update registration form UI



Closes #45"



git commit -m "fix(transcription): handle Azure service timeouts



- Add retry logic for failed API calls

- Improve error messages for users

- Log detailed error information



Fixes #67"

```

#### 5. Pull Request Process

**PR Checklist**
- [ ] Code follows style guidelines
- [ ] All tests pass
- [ ] Documentation updated
- [ ] Security considerations reviewed
- [ ] Performance impact assessed
- [ ] Breaking changes documented

**PR Template**
```markdown

## Description

Brief description of changes



## Type of Change

- [ ] Bug fix

- [ ] New feature

- [ ] Breaking change

- [ ] Documentation update



## Testing

- [ ] Unit tests added/updated

- [ ] Integration tests pass

- [ ] Manual testing completed



## Security

- [ ] No sensitive data exposed

- [ ] Input validation implemented

- [ ] Access controls maintained



## Performance

- [ ] No performance degradation

- [ ] Database queries optimized

- [ ] Resource usage considered

```

### Feature Development

#### Adding New Languages
```python

# 1. Update environment configuration

ALLOWED_LANGS = {

    "en-US": "English (United States)",

    "es-ES": "Spanish (Spain)",

    "new-LANG": "New Language Name"

}



# 2. Test language support

def test_new_language():

    # Verify Azure Speech Services supports the language

    # Test transcription accuracy

    # Update documentation

```

#### Adding New Audio Formats
```python

# 1. Update supported formats list

AUDIO_FORMATS = [

    "wav", "mp3", "ogg", "opus", "flac", 

    "new_format"  # Add new format

]



# 2. Update FFmpeg conversion logic

def _convert_to_audio(self, input_path, output_path, audio_format="wav"):

    if audio_format == "new_format":

        # Add specific conversion parameters

        cmd = ["ffmpeg", "-i", input_path, "-codec", "new_codec", output_path]

```

#### Adding New Features
```python

# 1. Database schema updates

def upgrade_database_schema():

    with self.get_connection() as conn:

        conn.execute("""

            ALTER TABLE transcriptions 

            ADD COLUMN new_feature_data TEXT

        """)



# 2. API endpoint updates

def new_feature_endpoint(user_id: str, feature_data: Dict) -> Dict:

    # Implement new feature logic

    pass



# 3. UI updates

def add_new_feature_ui():

    new_feature_input = gr.Textbox(label="New Feature")

    new_feature_button = gr.Button("Use New Feature")

```

---

## ⚙️ Advanced Configuration

### Performance Optimization

#### Concurrent Processing
```python

# Adjust worker thread pool size based on server capacity

class TranscriptionManager:

    def __init__(self, max_workers: int = None):

        if max_workers is None:

            # Auto-detect based on CPU cores

            import multiprocessing

            max_workers = min(multiprocessing.cpu_count(), 10)

        

        self.executor = ThreadPoolExecutor(max_workers=max_workers)



# Configure based on server specs

# Small server: max_workers=2-4

# Medium server: max_workers=5-8  

# Large server: max_workers=10+

```

#### Database Optimization
```python

# SQLite performance tuning

def configure_database_performance(db_path: str):

    with sqlite3.connect(db_path) as conn:

        # Enable WAL mode for better concurrency

        conn.execute("PRAGMA journal_mode=WAL")

        

        # Increase cache size (in KB)

        conn.execute("PRAGMA cache_size=10000")

        

        # Optimize synchronization

        conn.execute("PRAGMA synchronous=NORMAL")

        

        # Enable foreign keys

        conn.execute("PRAGMA foreign_keys=ON")

```

#### Memory Management
```python

# Large file handling

def process_large_file(file_path: str):

    """Process large files in chunks to manage memory"""

    chunk_size = 64 * 1024 * 1024  # 64MB chunks

    

    with open(file_path, 'rb') as f:

        while chunk := f.read(chunk_size):

            # Process chunk

            yield chunk



# Garbage collection for long-running processes

import gc



def cleanup_memory():

    """Force garbage collection"""

    gc.collect()

    

# Schedule periodic cleanup

schedule.every(30).minutes.do(cleanup_memory)

```

### Security Hardening

#### Rate Limiting
```python

from collections import defaultdict

from time import time



class RateLimiter:

    def __init__(self, max_requests: int = 100, window: int = 3600):

        self.max_requests = max_requests

        self.window = window

        self.requests = defaultdict(list)

    

    def is_allowed(self, user_id: str) -> bool:

        now = time()

        user_requests = self.requests[user_id]

        

        # Clean old requests

        user_requests[:] = [req_time for req_time in user_requests 

                           if now - req_time < self.window]

        

        # Check limit

        if len(user_requests) >= self.max_requests:

            return False

        

        user_requests.append(now)

        return True



# Usage in endpoints

rate_limiter = RateLimiter(max_requests=50, window=3600)  # 50 per hour



def submit_transcription(self, user_id: str, ...):

    if not rate_limiter.is_allowed(user_id):

        raise Exception("Rate limit exceeded")

```

#### Input Sanitization
```python

import bleach

import re



def sanitize_filename(filename: str) -> str:

    """Sanitize uploaded filename"""

    # Remove path traversal attempts

    filename = os.path.basename(filename)

    

    # Remove dangerous characters

    filename = re.sub(r'[<>:"/\\|?*]', '_', filename)

    

    # Limit length

    if len(filename) > 255:

        name, ext = os.path.splitext(filename)

        filename = name[:250] + ext

    

    return filename



def sanitize_user_input(text: str) -> str:

    """Sanitize user text input"""

    # Remove HTML tags

    text = bleach.clean(text, tags=[], strip=True)

    

    # Limit length

    text = text[:1000]

    

    return text.strip()

```

#### Audit Logging
```python

class AuditLogger:

    def __init__(self):

        self.logger = logging.getLogger('audit')

    

    def log_user_action(self, user_id: str, action: str, details: Dict = None):

        """Log user actions for security auditing"""

        audit_entry = {

            'timestamp': datetime.now().isoformat(),

            'user_id': user_id,

            'action': action,

            'details': details or {},

            'ip_address': self._get_client_ip(),

            'user_agent': self._get_user_agent()

        }

        

        self.logger.info(json.dumps(audit_entry))

    

    def _get_client_ip(self) -> str:

        # Implementation depends on deployment setup

        return "unknown"

    

    def _get_user_agent(self) -> str:

        # Implementation depends on deployment setup

        return "unknown"



# Usage

audit = AuditLogger()

audit.log_user_action(user_id, "login", {"success": True})

audit.log_user_action(user_id, "transcription_submit", {"filename": filename})

```

### Custom Extensions

#### Plugin Architecture
```python

class TranscriptionPlugin:

    """Base class for transcription plugins"""

    

    def pre_process(self, file_bytes: bytes, settings: Dict) -> bytes:

        """Pre-process audio before transcription"""

        return file_bytes

    

    def post_process(self, transcript: str, settings: Dict) -> str:

        """Post-process transcript text"""

        return transcript

    

    def get_name(self) -> str:

        """Return plugin name"""

        raise NotImplementedError



class NoiseReductionPlugin(TranscriptionPlugin):

    def get_name(self) -> str:

        return "noise_reduction"

    

    def pre_process(self, file_bytes: bytes, settings: Dict) -> bytes:

        # Implement noise reduction using audio processing library

        # This is a placeholder - actual implementation would use

        # libraries like librosa, scipy, or pydub

        return file_bytes



class LanguageDetectionPlugin(TranscriptionPlugin):

    def get_name(self) -> str:

        return "language_detection"

    

    def pre_process(self, file_bytes: bytes, settings: Dict) -> bytes:

        # Detect language and update settings

        detected_language = self._detect_language(file_bytes)

        settings['detected_language'] = detected_language

        return file_bytes



# Plugin manager

class PluginManager:

    def __init__(self):

        self.plugins: List[TranscriptionPlugin] = []

    

    def register_plugin(self, plugin: TranscriptionPlugin):

        self.plugins.append(plugin)

    

    def apply_pre_processing(self, file_bytes: bytes, settings: Dict) -> bytes:

        for plugin in self.plugins:

            file_bytes = plugin.pre_process(file_bytes, settings)

        return file_bytes

    

    def apply_post_processing(self, transcript: str, settings: Dict) -> str:

        for plugin in self.plugins:

            transcript = plugin.post_process(transcript, settings)

        return transcript

```

---

## 🔧 Troubleshooting

### Common Development Issues

#### Environment Setup Problems

**Issue**: Azure connection fails
```bash

# Check environment variables

python -c "

import os

print('AZURE_SPEECH_KEY:', bool(os.getenv('AZURE_SPEECH_KEY')))

print('AZURE_BLOB_CONNECTION:', bool(os.getenv('AZURE_BLOB_CONNECTION')))

"



# Test Azure connection

python -c "

from azure.storage.blob import BlobServiceClient

client = BlobServiceClient.from_connection_string('$AZURE_BLOB_CONNECTION')

print('Containers:', list(client.list_containers()))

"

```

**Issue**: FFmpeg not found
```bash

# Check FFmpeg installation

ffmpeg -version



# Install FFmpeg (Ubuntu/Debian)

sudo apt update && sudo apt install ffmpeg



# Install FFmpeg (Windows with Chocolatey)

choco install ffmpeg



# Install FFmpeg (macOS with Homebrew)

brew install ffmpeg

```

**Issue**: Database initialization fails
```python

# Check database permissions

import os

db_dir = "database"

if not os.path.exists(db_dir):

    os.makedirs(db_dir)

    print(f"Created directory: {db_dir}")



# Test database creation

import sqlite3

conn = sqlite3.connect("database/test.db")

conn.execute("CREATE TABLE test (id INTEGER)")

conn.close()

print("Database test successful")

```

#### Runtime Issues

**Issue**: Memory errors with large files
```python

# Monitor memory usage

import psutil



def check_memory():

    memory = psutil.virtual_memory()

    print(f"Memory usage: {memory.percent}%")

    print(f"Available: {memory.available / 1024**3:.1f}GB")



# Implement file chunking for large uploads

def process_large_file_in_chunks(file_path: str, chunk_size: int = 64*1024*1024):

    with open(file_path, 'rb') as f:

        while chunk := f.read(chunk_size):

            yield chunk

```

**Issue**: Transcription jobs stuck
```python

# Check pending jobs

def diagnose_stuck_jobs():

    pending_jobs = transcription_manager.db.get_pending_jobs()

    print(f"Pending jobs: {len(pending_jobs)}")

    

    for job in pending_jobs:

        duration = datetime.now() - datetime.fromisoformat(job.created_at)

        print(f"Job {job.job_id}: {job.status} for {duration}")

        

        if duration.total_seconds() > 3600:  # 1 hour

            print(f"⚠️ Job {job.job_id} may be stuck")



# Reset stuck jobs

def reset_stuck_jobs():

    with transcription_manager.db.get_connection() as conn:

        conn.execute("""

            UPDATE transcriptions 

            SET status = 'pending', azure_trans_id = NULL

            WHERE status = 'processing' 

            AND created_at < datetime('now', '-1 hour')

        """)

```

**Issue**: Azure API errors
```python

# Test Azure Speech Service

def test_azure_speech():

    try:

        url = f"{AZURE_SPEECH_KEY_ENDPOINT}/speechtotext/v3.2/transcriptions"

        headers = {"Ocp-Apim-Subscription-Key": AZURE_SPEECH_KEY}

        

        response = requests.get(url, headers=headers)

        print(f"Status: {response.status_code}")

        print(f"Response: {response.text[:200]}")

        

    except Exception as e:

        print(f"Azure Speech test failed: {e}")



# Check Azure service status

def check_azure_status():

    # Check Azure status page

    status_url = "https://status.azure.com/en-us/status"

    print(f"Check Azure status: {status_url}")

```

### Debugging Tools

#### Debug Mode Configuration
```python

# Enable debug mode

DEBUG = os.environ.get("DEBUG", "false").lower() == "true"



if DEBUG:

    logging.basicConfig(level=logging.DEBUG)

    

    # Enable Gradio debug mode

    demo.launch(debug=True, show_error=True)

```

#### Performance Profiling
```python

import cProfile

import pstats



def profile_function(func):

    """Profile function performance"""

    profiler = cProfile.Profile()

    

    def wrapper(*args, **kwargs):

        profiler.enable()

        result = func(*args, **kwargs)

        profiler.disable()

        

        # Print stats

        stats = pstats.Stats(profiler)

        stats.sort_stats('cumulative')

        stats.print_stats(10)  # Top 10 functions

        

        return result

    

    return wrapper



# Usage

@profile_function

def submit_transcription(self, ...):

    # Function implementation

    pass

```

#### Log Analysis
```python

def analyze_logs(log_file: str = "logs/transcription.log"):

    """Analyze application logs for issues"""

    

    errors = []

    warnings = []

    performance_issues = []

    

    with open(log_file, 'r') as f:

        for line in f:

            if 'ERROR' in line:

                errors.append(line.strip())

            elif 'WARNING' in line:

                warnings.append(line.strip())

            elif 'completed in' in line:

                # Extract timing information

                import re

                match = re.search(r'completed in (\d+\.\d+)s', line)

                if match and float(match.group(1)) > 30:  # > 30 seconds

                    performance_issues.append(line.strip())

    

    print(f"Errors: {len(errors)}")

    print(f"Warnings: {len(warnings)}")

    print(f"Performance issues: {len(performance_issues)}")

    

    return {

        'errors': errors[-10:],  # Last 10 errors

        'warnings': warnings[-10:],  # Last 10 warnings

        'performance_issues': performance_issues[-10:]

    }

```

### Production Troubleshooting

#### Service Health Check
```bash

#!/bin/bash

# health_check.sh



echo "=== System Health Check ==="



# Check service status

systemctl is-active transcription

systemctl is-active nginx



# Check disk space

df -h



# Check memory usage

free -h



# Check CPU usage

top -b -n1 | grep "Cpu(s)"



# Check logs for errors

tail -n 50 /home/transcription/app/logs/transcription.log | grep ERROR



# Check Azure connectivity

curl -s -o /dev/null -w "%{http_code}" https://azure.microsoft.com/



echo "=== Health Check Complete ==="

```

#### Database Recovery
```python

def recover_database():

    """Recover database from Azure backup"""

    try:

        # List available backups

        container_client = blob_service.get_container_client(AZURE_CONTAINER)

        backups = []

        

        for blob in container_client.list_blobs(name_starts_with="shared/backups/"):

            backups.append({

                'name': blob.name,

                'modified': blob.last_modified

            })

        

        # Sort by date (newest first)

        backups.sort(key=lambda x: x['modified'], reverse=True)

        

        if not backups:

            print("No backups found")

            return

        

        # Download latest backup

        latest_backup = backups[0]['name']

        print(f"Restoring from: {latest_backup}")

        

        blob_client = blob_service.get_blob_client(

            container=AZURE_CONTAINER,

            blob=latest_backup

        )

        

        # Download backup

        with open("database/transcriptions_restored.db", "wb") as f:

            f.write(blob_client.download_blob().readall())

        

        print("Database restored successfully")

        print("Restart the application to use restored database")

        

    except Exception as e:

        print(f"Database recovery failed: {str(e)}")

```

---

## 📚 Additional Resources

### Documentation Links
- [Azure Speech Services Documentation](https://docs.microsoft.com/en-us/azure/cognitive-services/speech-service/)
- [Azure Blob Storage Documentation](https://docs.microsoft.com/en-us/azure/storage/blobs/)
- [Gradio Documentation](https://gradio.app/docs/)
- [SQLite Documentation](https://www.sqlite.org/docs.html)
- [FFmpeg Documentation](https://ffmpeg.org/documentation.html)

### Useful Tools
- **Azure Storage Explorer**: GUI for managing blob storage
- **DB Browser for SQLite**: Visual database management
- **Postman**: API testing and development
- **Azure CLI**: Command-line Azure management
- **Visual Studio Code**: Recommended IDE with Azure extensions

### Community Resources
- [Azure Speech Services Community](https://docs.microsoft.com/en-us/answers/topics/azure-speech-services.html)
- [Gradio Community](https://github.com/gradio-app/gradio/discussions)
- [Python Audio Processing Libraries](https://github.com/topics/audio-processing)

---

**This developer guide provides comprehensive information for setting up, developing, deploying, and maintaining the Azure Speech Transcription service. For additional help, refer to the linked documentation and community resources.** 🚀