# -*- coding: utf-8 -*-
# app.py Blockchain Traceability Web Application
# Author: CIBA IoT Traceability Project
# Version: 2025.10 (Robust URL & JSON Handling)
from flask import Flask, render_template, request, jsonify, redirect, url_for as flask_url_for, send_from_directory, abort
import hashlib
import json
import datetime
from io import BytesIO
import base64
import os
import csv
import re
import ast
from urllib.parse import unquote_plus, urlencode
import pandas as pd
from datetime import datetime
from urllib.parse import unquote, quote
from urllib.parse import unquote
import csv
from datetime import datetime
from flask import current_app
from config import PUBLIC_BASE_URL, PUBLIC_PATH, FLASK_HOST, FLASK_PORT, public_url
# --- Import Blockchain Logic and Lineages ---
from blockchain_logic import (
Blockchain, find_block_by_data_id, get_batch_id_from_block,
STAGES, LINEAGE_P_VANNAMEI, LINEAGE_P_MONODON, LINEAGE_P_INDICUS,
LINEAGE_XLSX_2025
)
app = Flask(__name__, template_folder='templates', static_folder='static')
# Anchor every "static/..." lookup to this file's own directory rather than
# the process's current working directory. Under `python app.py` or Gunicorn
# (started with the project dir as cwd) relative paths like
# os.path.join('static', 'json', 'block_mapping.json') happen to resolve
# correctly, but under mod_wsgi the Apache worker's cwd is NOT the app
# directory (it's typically "/" unless WSGIDaemonProcess sets home=...), so
# those same relative paths silently fail with e.g.
# "Mapping file missing at static/json/block_mapping.json" in production.
# blockchain_logic.py already anchors its .db paths the same way for the
# same reason (see BASE_DIR there) -- this mirrors that fix for app.py.
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# Static public links/QR mapping. Customer-facing URLs are pre-generated and
# stored in static/json/static_links.json. No customer URL is constructed at
# request time.
STATIC_LINKS_PATH = os.path.join(os.path.dirname(__file__), "static", "json", "static_links.json")
with open(STATIC_LINKS_PATH, "r", encoding="utf-8") as _f:
STATIC_LINKS = json.load(_f)
class PublicPrefixMiddleware:
"""Accept requests both with and without the public mount prefix.
This makes deployment tolerant of either a reverse proxy that strips
/laksyashrimp/ or one that forwards the prefix unchanged.
"""
def __init__(self, application, prefix):
self.application = application
self.prefix = prefix.rstrip("/") or "/"
def __call__(self, environ, start_response):
path = environ.get("PATH_INFO", "") or "/"
if self.prefix != "/" and (path == self.prefix or path.startswith(self.prefix + "/")):
stripped = path[len(self.prefix):] or "/"
environ["PATH_INFO"] = stripped
# Keep SCRIPT_NAME consistent for WSGI-aware middleware/apps.
environ["SCRIPT_NAME"] = self.prefix
return self.application(environ, start_response)
app.wsgi_app = PublicPrefixMiddleware(app.wsgi_app, PUBLIC_PATH)
# --- Initialize all Blockchain Instances ---
BLOCKCHAINS = {
'vannamei': {
'instance': Blockchain(
db_name='blockchain_p_vannamei.db',
sample_lineage=LINEAGE_P_VANNAMEI
),
'name': 'TraceVannamei',
},
'monodon': {
'instance': Blockchain(
db_name='blockchain_p_monodon.db',
sample_lineage=LINEAGE_P_MONODON
),
'name': 'TraceMonodon',
},
'indicus': {
'instance': Blockchain(
db_name='blockchain_p_indicus.db',
sample_lineage=LINEAGE_P_INDICUS
),
'name': 'TraceIndicus',
},
'vannamei_2025': {
'instance': Blockchain(
db_name='blockchain_p_vannamei_2025.db',
sample_lineage=LINEAGE_XLSX_2025
),
'name': 'TraceVannamei2025',
},
}
# ==============================================================#
# -------- SUB-BLOCK LABELS FOR TraceVannamei2025 -------------#
# ==============================================================#
# Maps Unique ID -> (sub_label, block_group_number)
# Block 1 = Broodstock Facility (1a-1e)
# Block 2 = Hatchery-Nursery (2e only; 2a-2d not available)
# Block 3 = Grow-out System (3a-3e)
# Block 4 = Production of feed (4a-4e)
VANNAMEI_2025_SUB_LABELS = {
"3011231150A7010125": ("1a", 1),
"3011231015A2010125": ("1b", 1),
"3011231150A5010125": ("1c", 1),
"3011231150A4010125": ("1d", 1),
"301123100720010225": ("1e", 1),
"TNII15981720010127": ("2e", 2),
"TNII1598115001A726": ("3a", 3),
"TNII1598101501A226": ("3b", 3),
"TNII1598115001A526": ("3c", 3),
"TNII1598115001A426": ("3d", 3),
"TNII15981720010126_grow": ("3e", 3), # RGCA grow-out (3e)
"B101-4a": ("4a", 4),
"B101-4b": ("4b", 4),
"B101-4c": ("4c", 4),
"B101-4d": ("4d", 4),
"B102-4e": ("4e", 4),
}
# ==============================================================#
# ---------------------- HELPER FUNCTIONS ----------------------#
# ==============================================================#
# QR images are pre-generated static assets under static/qr.
# Customer-facing QR targets are stored in static/json/static_links.json.
def _try_parse_string_to_obj(s):
"""Try to convert a string that represents a dict/list into a Python object.
Tries json.loads first, then ast.literal_eval as a fallback.
Returns original value if parsing fails.
"""
if not isinstance(s, str):
return s
s_stripped = s.strip()
if not s_stripped:
return s
# Try JSON first (valid JSON uses double quotes)
try:
return json.loads(s_stripped)
except Exception:
pass
# Try ast.literal_eval for Python-style dict/list strings (single quotes etc)
try:
return ast.literal_eval(s_stripped)
except Exception:
pass
# fallback: try to fix common issues (replace smart quotes etc)
try:
# replace Windows newlines and weird quotes, then try json
fixed = s_stripped.replace("'", '"').replace("\\n", "\\n")
return json.loads(fixed)
except Exception:
# couldn't parse — return original string
return s
def public_url_for(endpoint, **values):
"""Jinja url_for that emits the public /laksyashrimp URL."""
return public_url(flask_url_for(endpoint, **values))
app.jinja_env.globals["url_for"] = public_url_for
app.jinja_env.globals["PUBLIC_BASE_URL"] = PUBLIC_BASE_URL
app.jinja_env.globals["PUBLIC_PATH"] = PUBLIC_PATH
app.jinja_env.globals["public_url"] = public_url
app.jinja_env.globals["STATIC_LINKS"] = STATIC_LINKS
def format_stage_display(stage_name):
if not stage_name:
return stage_name
suffix_map = {'1': 'A', '2': 'B', '3': 'C'}
stage_map = {
'Grow-out System': 3,
'Production of feed & inputs': 4,
'Hatchery-Nursery': 2,
'Broodstock Facility': 1,
'Harvesting': 5,
'Aggregation/Trading at Farm Gate': 6,
'Transportation': 7,
'Marketing': 8,
'Processing (direct from farms)': 9,
'Exporting': 10,
'Retailing': 11,
'Shrimp on a plate': 12
}
for base_stage, block_no in stage_map.items():
match = re.match(rf"^{re.escape(base_stage)}[- ]?([1-3])?$",
stage_name.strip(), flags=re.IGNORECASE)
if match:
num = match.group(1) or ""
letter = suffix_map.get(num, "")
if letter:
return f"Block #{block_no}{letter} - {stage_name}"
else:
return f"Block #{block_no} - {stage_name}"
return stage_name
def format_blocks_for_template(blocks_list, product_id):
"""Attach feedback QR, readable stage info and rating info for template rendering."""
formatted = []
for idx, block in enumerate(blocks_list, start=1):
data_id = get_batch_id_from_block(block)
stage_name = block.get("stage", "UnknownStage")
# For vannamei_2025, keep stage name clean — sub-block label is handled separately
display_stage = stage_name if product_id == 'vannamei_2025' else format_stage_display(stage_name)
# Customer-facing links are STATIC. They were generated once from the
# project data and stored in static/json/static_links.json.
static_item = STATIC_LINKS.get("dashboard", {}).get(product_id, {}).get(str(idx), {})
block_view_url = static_item.get("block_url", "")
feedback_url = static_item.get("feedback_url", "")
block_qr_static = static_item.get("block_qr", "")
feedback_qr_static = static_item.get("feedback_qr", "")
# compute average rating for this block (product_id and data_id)
avg, count = compute_avg_rating(product_id, data_id)
if avg is None:
avg_rating_text = None
avg_rating_value = None
else:
# format to one decimal place and include count
avg_rating_value = round(avg, 2)
avg_rating_text = f"{avg_rating_value:.1f} ({count})"
# --- Sub-block label for TraceVannamei2025 ---
block_sub_label = None
block_group_no = None
if product_id == 'vannamei_2025':
uid = (block.get('transactions') or [{}])[0].get('data', {}).get('Unique ID', '') or data_id
# Handle TNII15981720010126 appearing in both Block 2 (Hatchery) and Block 3 (Grow-out)
if uid == 'TNII15981720010126' and 'Grow-out' in stage_name:
sub_info = VANNAMEI_2025_SUB_LABELS.get('TNII15981720010126_grow')
else:
sub_info = VANNAMEI_2025_SUB_LABELS.get(uid)
if sub_info:
block_sub_label, block_group_no = sub_info
formatted.append({
'block': {**block, 'stage': display_stage},
'qr_url': block_view_url,
# block_qr_static/feedback_qr_static are stored in static_links.json
# as bare "/static/..." paths (they also double as on-disk file
# paths for the QR generator). They must be passed through
# public_url() here so the ![]()
src includes the "/laksyashrimp"
# reverse-proxy mount prefix -- otherwise the browser requests
# them from the domain root and gets a 404 (broken image).
'qr_img': public_url(block_qr_static) if block_qr_static else block_qr_static,
'feedback_qr_url': feedback_url,
'feedback_qr_img': public_url(feedback_qr_static) if feedback_qr_static else feedback_qr_static,
'avg_rating_text': avg_rating_text,
'avg_rating_value': avg_rating_value,
'batch_id': data_id,
'block_sub_label': block_sub_label,
'block_group_no': block_group_no
})
return formatted
def normalize_stage_name(name: str) -> str:
if not name:
return ""
return name.strip().lower().replace('/', '-').replace('_', '-')
def compute_avg_rating(product_id: str, batch_id: str):
"""
Read feedback//_feedback.csv and compute average rating.
Returns (avg_float, count) or (None, 0) if no file or no valid ratings.
"""
if not product_id:
return None, 0
product_folder = os.path.join("feedback", str(product_id).lower())
csv_file = os.path.join(product_folder, f"{str(batch_id).upper()}_feedback.csv")
if not os.path.exists(csv_file):
return None, 0
try:
with open(csv_file, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
# find rating column name (flexible)
rating_col = None
for c in reader.fieldnames or []:
if 'rating' in c.lower():
rating_col = c
break
if not rating_col:
return None, 0
total = 0.0
count = 0
for row in reader:
val = (row.get(rating_col) or "").strip()
if not val:
continue
# If someone stored non-numeric like '5/5' or text, extract digits
try:
num = float(val)
except Exception:
# try to extract first number
m = re.search(r"(\d+(\.\d+)?)", val)
if m:
num = float(m.group(1))
else:
continue
total += num
count += 1
if count == 0:
return None, 0
avg = total / count
return avg, count
except Exception as e:
# If anything fails, avoid crashing app — return no rating
print(f"⚠️ compute_avg_rating error for {csv_file}: {e}")
return None, 0
def safe_filename_part(s: str) -> str:
return re.sub(r'[^a-zA-Z0-9_-]', '_', str(s))
def extract_batch_id_from_block(block_data):
"""
Extract batch_id/unique_id from various block structures.
Handles both direct blocks and nested TransactionData structures.
"""
if not isinstance(block_data, dict):
return None
# Try direct fields first
for key in ['parent_batch_id', 'Unique ID', 'Unique Blockchain ID', 'ID', 'id', 'batch_id']:
if key in block_data and block_data[key]:
return str(block_data[key])
# Try TransactionData nested structure
trans_data = block_data.get('TransactionData') or block_data.get('data')
if isinstance(trans_data, dict):
for key in ['parent_batch_id', 'Unique ID', 'Unique Blockchain ID', 'ID', 'id', 'batch_id']:
if key in trans_data and trans_data[key]:
return str(trans_data[key])
return None
# ==============================================================#
# --------------------------- ROUTES ---------------------------#
# ==============================================================#
@app.route('/', methods=['GET', 'POST'])
def index():
product_id = request.args.get('product_id', 'vannamei')
product_data = BLOCKCHAINS.get(product_id)
if not product_data:
return redirect(public_url(flask_url_for('index', product_id='vannamei')))
blockchain_instance = product_data['instance']
alert_message = request.args.get('alert_message')
if request.method == 'POST':
selected_stage = request.form.get('selected_stage')
custom_stage_name = request.form.get('custom_stage_name')
transaction_data_str = request.form.get('transaction_data')
stage = custom_stage_name if selected_stage == 'New Custom Stage' else selected_stage
if not stage or not transaction_data_str:
alert_message = "error:Missing required fields (Stage and JSON Data)."
else:
try:
transaction_data = json.loads(transaction_data_str)
if 'ID' not in transaction_data and 'Unique ID' not in transaction_data:
alert_message = "error:JSON must include 'ID' or 'Unique ID'."
else:
new_block_index, new_id = blockchain_instance.mine(stage, transaction_data)
if new_block_index:
alert_message = f"success:Block #{new_block_index} mined! Batch ID: {new_id}."
else:
alert_message = f"warning:Block ID {new_id} already exists."
except json.JSONDecodeError:
alert_message = "error:Invalid JSON format."
except Exception as e:
alert_message = f"error:{e}"
return redirect(public_url(flask_url_for('index', product_id=product_id, alert_message=alert_message)))
blocks_to_display = [b.to_dict(include_hash=True) for b in blockchain_instance.chain[1:]]
filtered_blocks = []
for b in blocks_to_display:
stage_name = b.get("stage", "").lower()
if (
("shrimp on a plate" in stage_name or "customer view" in stage_name)
and any(x in stage_name for x in ["-1", "-2", "-3"])
):
continue
filtered_blocks.append(b)
full_chain_history = format_blocks_for_template(filtered_blocks, product_id)
# "Dashboard ->" link: product_trace() needs a real batch_id (e.g. "VAN-CIBA-001"),
# not the species key. Point it at the first real, existing batch export for this
# species so the link actually resolves; if no static/data export exists for this
# species yet (currently vannamei_2025), omit the link instead of pointing it at a
# batch_id that is guaranteed to 404 into "Unknown Product".
dashboard_batch_id = None
species_data_dir = os.path.join(BASE_DIR, "static", "data", product_id)
if os.path.isdir(species_data_dir):
batch_files = sorted(f for f in os.listdir(species_data_dir) if f.endswith(".json"))
if batch_files:
dashboard_batch_id = os.path.splitext(batch_files[0])[0]
return render_template(
'index.html',
stages=STAGES,
full_chain_history=full_chain_history,
alert_message=alert_message,
product_options=BLOCKCHAINS,
selected_product_id=product_id,
current_product_name=product_data['name'],
dashboard_batch_id=dashboard_batch_id
)
@app.route("/product/")
def product_trace(batch_id):
"""Display a complete product lineage from the exported product JSON."""
# Product exports are stored by species, not directly under static/data.
data_root = os.path.join(BASE_DIR, "static", "data")
candidates = []
for species_dir in ("vannamei", "monodon", "indicus"):
candidates.append(os.path.join(data_root, species_dir, f"{batch_id}.json"))
# Include the 2025 Vannamei dataset if present.
candidates.append(os.path.join(data_root, "vannamei_2025", f"{batch_id}.json"))
json_path = next((p for p in candidates if os.path.exists(p)), None)
if not json_path:
return render_template("traceability_view_full.html", batch=None)
with open(json_path, "r", encoding="utf-8") as f:
batch_data = json.load(f)
return render_template("traceability_view_full.html", batch=batch_data)
@app.route("/stage//")
def stage_trace(batch_id, stage_name):
"""Display one stage from a product export.
This keeps the legacy /stage/... QR links functional while using the
current public CIBA deployment.
"""
data_root = os.path.join(BASE_DIR, "static", "data")
json_path = None
for species_dir in ("vannamei", "monodon", "indicus", "vannamei_2025"):
candidate = os.path.join(data_root, species_dir, f"{batch_id}.json")
if os.path.exists(candidate):
json_path = candidate
break
if not json_path:
return render_template("trace_stage.html", batch_id=batch_id, stage_name=stage_name, entry=None), 404
with open(json_path, "r", encoding="utf-8") as f:
batch_data = json.load(f)
def norm(value):
return unquote_plus(str(value or "")).replace("+", " ").strip().lower()
requested = norm(stage_name)
entry = None
for candidate in batch_data.get("lineage", []):
candidate_stage = candidate.get("Stage") or candidate.get("stage") or ""
if norm(candidate_stage) == requested:
entry = candidate
break
return render_template(
"trace_stage.html",
batch_id=batch_id,
stage_name=stage_name,
entry=entry,
), (200 if entry else 404)
@app.route("/block//")
def view_block(product_id, block_number):
"""Display a single block (or entries for the block number) for a product.
This version normalizes transaction data strings into dicts for template rendering.
"""
# determine lineage_key
lineage_key = None
for key in BLOCKCHAINS.keys():
if key.lower() in product_id.lower() or product_id.lower() in key.lower():
lineage_key = key
break
if not lineage_key:
return f"Unknown lineage for {product_id}", 404
chain_instance = BLOCKCHAINS[lineage_key]["instance"]
# try to get chain data (list of dict-like blocks)
try:
chain_data = chain_instance.load_chain_for_product(product_id) or []
except Exception:
chain_data = [b.to_dict(include_hash=True) for b in chain_instance.chain[1:]]
if not chain_data:
return f"No data found for {product_id}", 404
# find matching blocks by block_number or batch_id
batch_id = request.args.get("batch_id")
matched = []
if batch_id:
for blk in chain_data:
try:
if get_batch_id_from_block(blk) == batch_id:
matched.append(blk)
except Exception:
if blk.get("ID") == batch_id or blk.get("id") == batch_id:
matched.append(blk)
if not matched:
needle = f"Block #{block_number}"
for blk in chain_data:
stage_field = (blk.get("DisplayName") or blk.get("Stage") or blk.get("stage") or "")
if needle in str(stage_field):
matched.append(blk)
if not matched:
# try numeric block number keys
for blk in chain_data:
if str(blk.get("BlockNumber") or blk.get("block_number") or blk.get("index") or "") == str(block_number):
matched.append(blk)
if not matched:
return f"No entries found for Block #{block_number}", 404
# Normalize transaction fields for each matched block
normalized_blocks = []
for blk in matched:
# ensure blk is a dict (if it's an object, try to convert)
if not isinstance(blk, dict):
try:
blk = blk.to_dict(include_hash=True)
except Exception:
# leave as-is
pass
# Look for common transaction containers and parse them if they are strings
possible_fields = ['data', 'Data', 'transaction_data', 'TransactionData', 'payload', 'payload_data']
normalized_data = None
for key in possible_fields:
if key in blk and blk[key]:
normalized_data = _try_parse_string_to_obj(blk[key])
# store normalized into a standard key 'data' for templates
blk['data'] = normalized_data
break
# If no transaction container found, but 'block' key present with nested data:
if 'data' not in blk and 'block' in blk and isinstance(blk['block'], dict):
for key in possible_fields:
if key in blk['block'] and blk['block'][key]:
blk['block']['data'] = _try_parse_string_to_obj(blk['block'][key])
break
# If everything above failed, try to parse blk.get('data') if exists as string
if 'data' in blk and isinstance(blk['data'], str):
blk['data'] = _try_parse_string_to_obj(blk['data'])
if 'block' in blk and isinstance(blk['block'], dict) and isinstance(blk['block'].get('data'), str):
blk['block']['data'] = _try_parse_string_to_obj(blk['block']['data'])
# also compute a readable 'stage_full' field for the template to show
stage_val = (blk.get('stage') or blk.get('Stage') or (blk.get('block') or {}).get('stage') or (blk.get('block') or {}).get('Stage') or "")
blk['stage_full'] = stage_val
normalized_blocks.append(blk)
# Create formatted items using existing helper so we get qr_img and qr_url consistent with index
formatted = format_blocks_for_template(normalized_blocks, product_id)
# choose first block to display (or adapt to show multiple)
block_to_show = formatted[0]
product_name = BLOCKCHAINS[lineage_key]["name"]
# determine batch_id to show
try:
batch_id_used = get_batch_id_from_block(normalized_blocks[0])
except Exception:
batch_id_used = batch_id or normalized_blocks[0].get('ID') or normalized_blocks[0].get('id') or ""
# compute a block index display (try several keys)
block_index = (normalized_blocks[0].get('index') or normalized_blocks[0].get('BlockNumber') or
normalized_blocks[0].get('block_number') or "")
return render_template(
"block_detail.html",
block=block_to_show,
product_name=product_name,
batch_id=batch_id_used,
block_index=block_index
)
# ==============================================================#
# ----------------- UNIVERSAL LINEAGE BLOCK ROUTE --------------#
# ==============================================================#
@app.route("/lineage_block//")
def lineage_block(species, block_number):
"""
Clean synchronized route always uses block_mapping.json for correct lookup.
Shows the correct block from /static/json// and renders lineage_block_view.html.
QR display removed.
"""
import json
species_key = species.lower()
mapping_path = os.path.join(BASE_DIR, 'static', 'json', 'block_mapping.json')
if species_key not in BLOCKCHAINS:
return f"Unknown species '{species}'", 404
# ? Load mapping
if not os.path.exists(mapping_path):
return f"Mapping file missing at {mapping_path}", 404
try:
with open(mapping_path, "r") as f:
mapping_data = json.load(f)
except Exception as e:
return f"Error reading mapping file: {e}", 500
species_map = mapping_data.get(species_key, {})
matched_uid = None
matched_filename = None
# Find the file with matching block index
for uid, info in species_map.items():
if info.get("index") == block_number:
matched_uid = uid
matched_filename = info.get("filename")
break
if not matched_filename:
return f"No block found for index {block_number}", 404
# ? Load JSON file for the correct block
json_path = os.path.join(BASE_DIR, "static", "json", species_key, matched_filename)
if not os.path.exists(json_path):
return f"JSON file not found: {matched_filename}", 404
try:
with open(json_path, "r", encoding="utf-8") as f:
block_data = json.load(f)
except Exception as e:
return f"Error reading JSON: {e}", 500
# Extract useful info
batch_id = extract_batch_id_from_block(block_data) or matched_uid
stage_name = block_data.get("Stage") or block_data.get("stage") or f"Block {block_number}"
# URLs (no QR now)
json_url = public_url(f"/static/json/{species_key}/{matched_filename}")
# Rating
avg, count = compute_avg_rating(species_key, batch_id)
avg_rating_text = f"{round(avg,2):.1f} ({count})" if avg is not None else None
# Sub-block label for vannamei_2025
block_sub_label = None
if species_key == 'vannamei_2025':
uid = extract_batch_id_from_block(block_data) or matched_uid
stage_chk = block_data.get('Stage') or block_data.get('stage') or ''
if uid == 'TNII15981720010126' and 'Grow-out' in stage_chk:
sub_info = VANNAMEI_2025_SUB_LABELS.get('TNII15981720010126_grow')
else:
sub_info = VANNAMEI_2025_SUB_LABELS.get(uid)
if sub_info:
block_sub_label = sub_info[0]
return render_template(
"lineage_block_view.html",
species=species_key,
block_number=block_number,
stage_name=stage_name,
block=block_data,
raw_block_data=block_data,
total_blocks=len(species_map),
feedback_url=None, # hide feedback button
qr_img=None, # no QR image
qr_img_url=None,
block_json_url=json_url, # link to JSON only
avg_rating_text=avg_rating_text,
avg_rating_value=avg,
qr_info={"species": species_key, "block_number": block_number, "source_file": matched_filename},
block_sub_label=block_sub_label,
)
# ==============================================================#
# ----------------------- FEEDBACK ROUTE -----------------------#
# ==============================================================#
@app.route("/localhost_feedback", methods=["GET"])
def localhost_feedback():
"""Display feedback form dynamically based on the detected block or stage."""
product_id = request.args.get("product_id", "")
batch_id = request.args.get("batch_id", "")
stage_raw = request.args.get("stage", "")
# Decode and normalize the stage safely
stage_decoded = unquote_plus(stage_raw).replace("_", " ").replace("%23", "#").strip()
stage_clean = stage_decoded.lower()
stage_display = stage_decoded or "Unknown Stage"
# Default feedback form
form_template = "feedback.html"
# --- FORM A: Blocks 1 3 (Broodstock, Hatchery, Nursery) ---
if any(x in stage_clean for x in [
"block #1", "block #2", "block #3a", "block #3b", "block #3c",
"broodstock", "hatchery", "nursery"
]) and not any(x in stage_clean for x in ["block #10", "block #11"]):
form_template = "feedback_form_a.html"
# --- FORM B: Block 4 (Grow-out) ---
elif any(x in stage_clean for x in ["block #4", "grow-out", "grow out"]):
form_template = "feedback_form_b.html"
# --- FORM C: Blocks 5 11 (Processing, Exporting, Marketing, etc.) ---
elif any(x in stage_clean for x in [
"block #5", "block #6", "block #7", "block #8", "block #9",
"block #10", "block #11", "processing", "exporting",
"retailing", "marketing"
]):
form_template = "feedback_form_c.html"
print(f"? Feedback opened for stage='{stage_display}' ? form={form_template}")
return render_template(
form_template,
product_id=product_id,
batch_id=batch_id,
stage_name=stage_display
)
@app.route("/submit_feedback", methods=["POST"])
def submit_feedback():
"""Save feedback data into species- and block-specific CSV files."""
name = request.form.get("name", "")
address = request.form.get("address", "")
contact_number = request.form.get("contact_number", "")
email_id = request.form.get("email_id", "")
freshness = request.form.get("freshness", "")
smell = request.form.get("smell", "")
appearance = request.form.get("appearance", "")
texture = request.form.get("texture", "")
black_gill = request.form.get("black_gill_black_spots", "")
shape_size = request.form.get("shape_size", "")
weight = request.form.get("weight", "")
rating = request.form.get("rating", "")
feedback_status = request.form.get("feedback_status", "")
remarks = request.form.get("remarks", "")
product_id = request.form.get("product_id", "").strip().lower()
batch_id = request.form.get("batch_id", "").strip().upper()
stage_name = request.form.get("stage_name", "").strip()
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 📁 Create species folder
base_folder = os.path.join("feedback", product_id)
os.makedirs(base_folder, exist_ok=True)
# 📝 Each block gets its own CSV file
feedback_file = os.path.join(base_folder, f"{batch_id}_feedback.csv")
# ✍️ Write header only if file doesn't exist
write_header = not os.path.exists(feedback_file)
with open(feedback_file, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if write_header:
writer.writerow([
"Timestamp", "Product ID", "Batch ID", "Stage Name",
"Name", "Address", "Contact Number", "Email ID",
"Freshness", "Smell", "Appearance", "Texture",
"Black Gill/Black Spots", "Shape & Size", "Weight",
"Rating (1-5)", "Feedback Status", "Remarks"
])
writer.writerow([
timestamp, product_id, batch_id, stage_name,
name, address, contact_number, email_id,
freshness, smell, appearance, texture,
black_gill, shape_size, weight,
rating, feedback_status, remarks
])
print(f"✅ Feedback saved to {feedback_file}")
return render_template(
"feedback_ack.html",
stage=stage_name,
product_id=product_id,
batch_id=batch_id,
rating=rating
)
# ==============================================================#
# --------------- BLOCK JSON VIEWING ROUTES --------------------#
# ==============================================================#
@app.route('/block_json//')
def view_block_json(species, filename):
"""
Display a single block's JSON data with QR code.
Species: vannamei, monodon, or indicus
Filename: the JSON filename (e.g., BATCH001.json)
"""
# Validate species
valid_species = list(BLOCKCHAINS.keys())
if species not in valid_species:
return f"Invalid species '{species}'", 404
# Check if JSON file exists
json_path = os.path.join(BASE_DIR, 'static', 'json', species, filename)
if not os.path.exists(json_path):
return f"Block JSON not found: {filename}", 404
# Check if QR code exists
qr_filename = filename.replace('.json', '.png')
qr_path = os.path.join(BASE_DIR, 'static', 'qrcodes', species, qr_filename)
qr_exists = os.path.exists(qr_path)
# Build URLs
json_url = public_url(f"/static/json/{species}/{filename}")
qr_url = public_url(f"/static/qrcodes/{species}/{qr_filename}") if qr_exists else ""
# Load the JSON data
try:
with open(json_path, 'r', encoding='utf-8') as f:
block_data = json.load(f)
except Exception as e:
return f"Error loading JSON: {e}", 500
return render_template('block_json_view.html',
species=species,
filename=filename,
json_url=json_url,
qr_url=qr_url,
block_data=block_data)
@app.route('/block_by_id//')
def view_block_by_id(species, block_id):
"""
View a block by its unique ID (batch_id, Unique ID, etc.)
This uses the block_mapping.json to find the correct block index
"""
species_key = species.lower()
if species_key not in BLOCKCHAINS:
return f"Unknown species '{species}'", 404
# Load block mapping
mapping_path = os.path.join(BASE_DIR, 'static', 'json', 'block_mapping.json')
if not os.path.exists(mapping_path):
# Fallback to lineage_block_mem with index 1
return redirect(public_url(flask_url_for('lineage_block', species=species_key, block_number=1)))
try:
with open(mapping_path, 'r') as f:
block_mapping = json.load(f)
except Exception as e:
print(f"ERROR: Could not load block mapping: {e}")
return f"Error loading block mapping: {e}", 500
# Find the block index for this unique ID
species_mapping = block_mapping.get(species_key, {})
if block_id not in species_mapping:
return f"Block ID '{block_id}' not found for {species}", 404
block_info = species_mapping[block_id]
block_index = block_info.get('index', 1)
# Redirect to the correct block by index
return redirect(public_url(flask_url_for('lineage_block', species=species_key, block_number=block_index)))
@app.route('/api/block_json//')
def get_block_json_api(species, filename):
"""API endpoint to fetch block JSON data"""
valid_species = list(BLOCKCHAINS.keys())
if species not in valid_species:
return jsonify({"error": "Invalid species"}), 404
json_path = os.path.join(BASE_DIR, 'static', 'json', species, filename)
if not os.path.exists(json_path):
return jsonify({"error": "Block not found"}), 404
try:
with open(json_path, 'r', encoding='utf-8') as f:
data = json.load(f)
return jsonify(data)
except Exception as e:
return jsonify({"error": str(e)}), 500
# ==============================================================#
# -------------------------- EXPORT CSV ------------------------#
# ==============================================================#
@app.route("/export_csv/")
def export_csv(species):
species = species.lower()
if species not in BLOCKCHAINS:
return jsonify({"error": f"Invalid species '{species}'"}), 400
csv_filename = f"{species}_blockchain_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
csv_path = os.path.join(BASE_DIR, "static", "exports", csv_filename)
os.makedirs(os.path.dirname(csv_path), exist_ok=True)
blockchain = BLOCKCHAINS[species]["instance"]
blocks = [b.to_dict(include_hash=True) for b in blockchain.chain[1:]]
with open(csv_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=blocks[0].keys())
writer.writeheader()
writer.writerows(blocks)
return jsonify({"message": "Export complete", "csv_path": csv_path})
# ==============================================================#
# ---------------- FEEDBACK COMPATIBILITY ROUTE ----------------#
# ==============================================================#
@app.route("/feedback", methods=["GET"])
def feedback_compatibility():
"""Compatibility alias for older /feedback links."""
return redirect(public_url(flask_url_for("localhost_feedback", **request.args.to_dict(flat=True))))
# ==============================================================#
# ----------- VANNAMEI 2025 LINEAGE TRACE PAGES ---------------#
# ==============================================================#
# Full lineage definitions for TraceVannamei2025
# Each lineage: list of (block_label, unique_id, stage, block_index_in_chain)
# block_index = position in chain (1-based, matching loop.index in template)
VANNAMEI_2025_LINEAGES = {
'a': {
'title': 'Lineage A',
'description': 'Broodstock 1a → (No Nursery) → Grow-out 3a → Feed 4a',
'steps': [
{'label': '1a', 'uid': '3011231150A7010125', 'stage': 'Broodstock Facility', 'chain_idx': 1},
{'label': '2a', 'uid': None, 'stage': 'Hatchery-Nursery', 'chain_idx': None, 'not_available': True},
{'label': '3a', 'uid': 'TNII1598115001A726', 'stage': 'Grow-out System', 'chain_idx': 7},
{'label': '4a', 'uid': 'B101-4a', 'stage': 'Production of feed & inputs','chain_idx': 12},
]
},
'b': {
'title': 'Lineage B',
'description': 'Broodstock 1b → (No Nursery) → Grow-out 3b → Feed 4b',
'steps': [
{'label': '1b', 'uid': '3011231015A2010125', 'stage': 'Broodstock Facility', 'chain_idx': 2},
{'label': '2b', 'uid': None, 'stage': 'Hatchery-Nursery', 'chain_idx': None, 'not_available': True},
{'label': '3b', 'uid': 'TNII1598101501A226', 'stage': 'Grow-out System', 'chain_idx': 8},
{'label': '4b', 'uid': 'B101-4b', 'stage': 'Production of feed & inputs','chain_idx': 13},
]
},
'c': {
'title': 'Lineage C',
'description': 'Broodstock 1c → (No Nursery) → Grow-out 3c → Feed 4c',
'steps': [
{'label': '1c', 'uid': '3011231150A5010125', 'stage': 'Broodstock Facility', 'chain_idx': 3},
{'label': '2c', 'uid': None, 'stage': 'Hatchery-Nursery', 'chain_idx': None, 'not_available': True},
{'label': '3c', 'uid': 'TNII1598115001A526', 'stage': 'Grow-out System', 'chain_idx': 9},
{'label': '4c', 'uid': 'B101-4c', 'stage': 'Production of feed & inputs','chain_idx': 14},
]
},
'd': {
'title': 'Lineage D',
'description': 'Broodstock 1d → (No Nursery) → Grow-out 3d → Feed 4d',
'steps': [
{'label': '1d', 'uid': '3011231150A4010125', 'stage': 'Broodstock Facility', 'chain_idx': 4},
{'label': '2d', 'uid': None, 'stage': 'Hatchery-Nursery', 'chain_idx': None, 'not_available': True},
{'label': '3d', 'uid': 'TNII1598115001A426', 'stage': 'Grow-out System', 'chain_idx': 10},
{'label': '4d', 'uid': 'B101-4d', 'stage': 'Production of feed & inputs','chain_idx': 15},
]
},
'e': {
'title': 'Lineage E',
'description': 'Broodstock 1e → Hatchery 2e → Grow-out 3e → Feed 4e',
'steps': [
{'label': '1e', 'uid': '301123100720010225', 'stage': 'Broodstock Facility', 'chain_idx': 5},
{'label': '2e', 'uid': 'TNII15981720010126', 'stage': 'Hatchery-Nursery', 'chain_idx': 6},
{'label': '3e', 'uid': 'TNII15981720010126_grow', 'stage': 'Grow-out System (RGCA)', 'chain_idx': 11},
{'label': '4e', 'uid': 'B102-4e', 'stage': 'Production of feed & inputs','chain_idx': 16},
]
},
}
@app.route('/lineage_trace/vannamei_2025/')
def vannamei_lineage_trace(lineage_id):
lineage_id = lineage_id.lower()
if lineage_id not in VANNAMEI_2025_LINEAGES:
return f"Unknown lineage '{lineage_id}'. Valid: a, b, c, d, e", 404
lineage = VANNAMEI_2025_LINEAGES[lineage_id]
bc = BLOCKCHAINS['vannamei_2025']['instance']
# Build enriched steps with block data + QR codes
enriched_steps = []
for step in lineage['steps']:
chain_idx = step.get('chain_idx')
not_available = step.get('not_available', False)
note = step.get('note', None)
block_data = None
block_qr_img = None
block_qr_url = None
feedback_qr_img = None
feedback_qr_url = None
if chain_idx and not not_available:
# Get block from chain
if chain_idx < len(bc.chain):
b = bc.chain[chain_idx]
block_data = b.transactions[0]['data'] if b.transactions else {}
# Static pre-generated URLs/QR files.
static_item = STATIC_LINKS.get("vannamei_2025_lineage", {}).get(lineage_id, {}).get(step["label"], {})
block_qr_url = static_item.get("block_url", "")
feedback_qr_url = static_item.get("feedback_url", "")
# Same fix as format_blocks_for_template(): these come out of
# static_links.json as bare "/static/..." paths and need the
# "/laksyashrimp" prefix added back for the browser to find them.
_raw_block_qr = static_item.get("block_qr", "")
_raw_feedback_qr = static_item.get("feedback_qr", "")
block_qr_img = public_url(_raw_block_qr) if _raw_block_qr else _raw_block_qr
feedback_qr_img = public_url(_raw_feedback_qr) if _raw_feedback_qr else _raw_feedback_qr
enriched_steps.append({
'label': step['label'],
'uid': step['uid'],
'stage': step['stage'],
'chain_idx': chain_idx,
'not_available': not_available,
'note': note,
'block_data': block_data,
'block_qr_img': block_qr_img,
'block_qr_url': block_qr_url,
'feedback_qr_img': feedback_qr_img,
'feedback_qr_url': feedback_qr_url,
})
return render_template(
'vannamei_lineage_trace.html',
lineage_id=lineage_id,
lineage=lineage,
steps=enriched_steps,
all_lineages=VANNAMEI_2025_LINEAGES,
)
# ==============================================================#
# -------------------------- MAIN RUN --------------------------#
# ==============================================================#
if __name__ == "__main__":
app.run(host=FLASK_HOST, port=FLASK_PORT, debug=False)