git/reactiveData
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
Backup user agent 2025-06-06 13:59:54 +04:00
commit f56f2c8338
12 changed files with 368 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
.env.sample.php Normal file
View File

@ -0,0 +1,23 @@
<?php
//
// 1. MariaDB Connection Credentials
//
define("DB_HOST", "localhost");
define("DB_USER", "SQL_DB_USER");
define("DB_PASS", "SQL_DB_PASS");
define("DB_NAME", "SQL_DB_NAME");
//
// 2. Telegram Reporting
//
$BOT_TOKEN = '1234567890:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'; // Change to YOUR Telegram Bot Token
$CHAT_IDS = [
"NAME_OF_CONTACT_OR_GROUP" => "12345678",
];

9
.gitignore vendored Normal file
View File

@ -0,0 +1,9 @@
vendor/
composer.lock
composer.sync
.env
.env.php
test_code.php

10
DbContinuousIntegration.php Executable file
View File

@ -0,0 +1,10 @@
<?php
require_once __DIR__."/.env.php";
require __DIR__."/app.php";
require __DIR__."/plt.php";
require __DIR__."/sys.php";
main();

9
DbContinuousIntegrationWrapper.sh Executable file
View File

@ -0,0 +1,9 @@
#!/bin/bash
cd "$(dirname ${BASH_SOURCE[0]})"
while [[ 1 ]]
do
killall DbContinuousIntegrationWrapper.sh
/usr/bin/php DbContinuousIntegration.php
make
sleep 10
done

10
Makefile Normal file
View File

@ -0,0 +1,10 @@
#!/bin/bash
composer.sync: composer.json
COMPOSER_ALLOW_SUPERUSER=1 composer update 2>&1 > $@
install:
echo "Add this line to crontab -e"
echo '@reboot /bin/bash /root/DbContinuousIntegration/DbContinuousIntegrationWrapper.sh 2> /dev/null > /dev/null &'
echo
echo 'CREATE TABLE SYS_PRD_BND.Tables (Name, onUpdate_phpCode, onUpdate_pyCode, LastUpdated);' | sudo mysql

23
app.php Executable file
View File

@ -0,0 +1,23 @@
<?php
/*
*
* 1. APP-Level
*
* */
/*
* Create a new table handler:
* (1) create a function called handleXXXXRow(data, error)
* (1.1) XXX is DbName__TableName
* (2) function should return "true" on success
* (3) function should fill in "error" on error and return false
*
*/
function main() {
processAllTheActiveTables();
sleep(1);
}

5
composer.json Normal file
View File

@ -0,0 +1,5 @@
{
"require": {
"denpa\/php-bitcoinrpc": "^2.2"
}
}

27
configure vendored Executable file
View File

@ -0,0 +1,27 @@
#!/bin/bash
# Prompt for user input with defaults
read -p "Enter DB Host [localhost]: " DB_HOST
DB_HOST=${DB_HOST:-localhost}
read -p "Enter DB User [user]: " DB_USER
DB_USER=${DB_USER:-user}
read -p "Enter DB Password [pass]: " DB_PASS
DB_PASS=${DB_PASS:-pass}
read -p "Enter DB Name [database]: " DB_NAME
DB_NAME=${DB_NAME:-database}
# Generate .env.php file
cat <<EOL > .env.php
<?php
define("DB_HOST", "$DB_HOST");
define("DB_USER", "$DB_USER");
define("DB_PASS", "$DB_PASS");
define("DB_NAME", "$DB_NAME");
EOL
echo ".env.php created successfully."

10
install/SYS_PRD_BND.Composer.sql Normal file
View File

@ -0,0 +1,10 @@
USE SYS_PRD_BND;
CREATE TABLE `Composer` (
`VendorName` varchar(255) NOT NULL,
`PackageName` varchar(255) NOT NULL,
`VersionOrBranch` varchar(25) NOT NULL,
`RepositoryUrl` varchar(255) DEFAULT NULL,
`RepositoryType` enum('vcs') NOT NULL DEFAULT 'vcs',
`LastUpdated` timestamp NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
PRIMARY KEY (`VendorName`,`PackageName`)
);

9
install/SYS_PRD_BND.Tables.sql Normal file
View File

@ -0,0 +1,9 @@
USE SYS_PRD_BND;
CREATE TABLE `Tables` (
`Name` varchar(255) NOT NULL,
`onUpdate_phpCode` text DEFAULT NULL,
`onUpdate_pyCode` text DEFAULT NULL,
`LastError` text DEFAULT NULL,
`LastUpdated` timestamp NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
PRIMARY KEY (`Name`)
);

165
plt.php Executable file
View File

@ -0,0 +1,165 @@
<?php
/*
*
* 2. PLT-Level
*
*
*/
function processAllTheActiveTables() {
echo "Scanning all the tables that need updating...\n";
foreach(sql("SELECT Name, onUpdate_phpCode, onUpdate_pyCode, LastUpdated FROM SYS_PRD_BND.Tables") as $activeTable) {
extract($activeTable);
echo "Found Table : ".greenText($Name)."\n";
echo "Scanning all the rows in table $Name that need to be ran through trigger code:\n";
foreach(sql("SELECT * FROM $Name WHERE LastUpdated > '$LastUpdated'") as $unprocessedRow) {
echo "Found row\n".json_encode($unprocessedRow)."\n";
$functionName = "handleNew".str_replace(".","__",$activeTable["Name"])."Row";
echo "Checking if function ".greenText($functionName)." exists at code-level (sys).\n";
if (function_exists($functionName)) {
echo "Calling function ".greenText("$functionName(...)")."\n";
if ($functionName($unprocessedRow, $error) === false) {
sql("UPDATE SYS_PRD_BND.Tables SET LastError = '".str_replace("'",'"',$error)."', LastUpdated = NOW() WHERE Name = '{$activeTable["Name"]}'");
sendTelegramMessage("*ERROR* on trigger function for table {$activeTable["Name"]} : ".$error,"DatabaseGroup");
} else {
sql("UPDATE SYS_PRD_BND.Tables SET LastError = '', LastUpdated = NOW() WHERE Name = '{$activeTable["Name"]}'");
}
}
/*
* PHP CODE
*/
echo "Checking if function ".greenText($functionName)." exists in PHP at database-level (app).\n";
if (!is_null($onUpdate_phpCode) && !empty($onUpdate_phpCode)) {
echo "Creating function ".greenText("$functionName(...)")." context\n";
// 1. Add the PHP START TAG
$code = "<"."?"."php \n";
// 2. Add the CONSTANTS
foreach(sql("SELECT Name, Type, Value FROM SYS_PRD_BND.Constants") as $const)
$code .= "define(\"{$const["Name"]}\",".($const["Type"]!="String"?$const["Value"]:'"'.$const["Value"].'"').");\n";
// 3. Add the SUPPORT FUNCTIONS
foreach(sql("SELECT Name, InputArgs_json, PhpCode FROM SYS_PRD_BND.SupportFunctions WHERE PhpCode IS NOT NULL") as $f)
$code .= "function {$f["Name"]} (".implode(", ",array_map(fn($s)=>"\$$s",array_keys(json_decode($f["InputArgs_json"],1)))).") {\n".$f["PhpCode"]."\n}\n";
$code .= "require_once '/root/DbContinuousIntegration/sys.php'; \n";
$code .= "function $functionName (&\$data, &\$error) {\n";
$code .= $onUpdate_phpCode;
$code .= "\n}\n";
$code .= "\$data = ".var_export($unprocessedRow,1).";\n";
$code .= "\$initial_data = json_encode(".var_export($unprocessedRow,1).");\n";
$code .= "$functionName(\$data,\$error);";
$code .= "\necho json_encode(\$data);\n";
file_put_contents(__DIR__."/test_code.php", $code);
echo "Running function ".greenText("$functionName(...)")." in sandbox environment\n";
if (runProcess("/usr/bin/php",$code,$stdout,$error) != 0) {
sql("UPDATE SYS_PRD_BND.Tables SET LastError = '".str_replace("'",'"',$error)."', LastUpdated = NOW() WHERE Name = '{$activeTable["Name"]}'");
sendTelegramMessage("*ERROR* on trigger function for table {$activeTable["Name"]} : ".$error,"DatabaseGroup");
} else {
// Update database row if needed
$newRowValue = json_decode($stdout,1);
echo "\n".redText("DEBUG new-row-value-json: ").json_encode($newRowValue)."\n";
echo "\n".redText("DEBUG unprocessedRow-json: ").json_encode($unprocessedRow)."\n";
if (json_encode($newRowValue) != json_encode($unprocessedRow)) {
$pkColsName = getTblPrimaryKeyColName($activeTable["Name"]);
$pkColsValues = array_map(fn($cName)=>$unprocessedRow[$cName],$pkColsName);
$sql_instruction = "";
$sql_instruction .= "UPDATE ";
$sql_instruction .= $activeTable["Name"];
$sql_instruction .= " SET ".implode(",",array_map(fn($k,$v)=>"$k=".(is_numeric($v)?$v:"'".str_replace("'","''",$v)."'"),array_keys($newRowValue),array_values($newRowValue)));
$sql_instruction .= " WHERE ".implode(" AND ",array_map(fn($k,$v)=>"$k = ".(is_numeric($newRowValue[$k])?$newRowValue[$k]:'"'.$newRowValue[$k].'"'),$pkColsName,$pkColsValues));
echo "\n".redText("DEBUG sql-instruction: ").$sql_instruction."\n";
sql($sql_instruction);
}
// Update the status of the operation
sql("UPDATE SYS_PRD_BND.Tables SET LastError = '', LastUpdated = NOW() WHERE Name = '{$activeTable["Name"]}'");
}
}
/*
* Python CODE
*/
echo "Checking if function ".greenText($functionName)." exists in Python at database-level (app).\n";
if (!is_null($onUpdate_pyCode) && !empty($onUpdate_pyCode)) {
echo "Creating function ".greenText("$functionName(...)")." context\n";
$code = "";
// 1. Insert the import to all the external libraries
foreach(sql("SELECT LibName, AliasName FROM SYS_PRD_BND.PyPi") as $module)
$code .= "import ".$module["LibName"] .(is_null($module["AliasName"])&&!empty($module["AliasName"])?" as ".$module["AliasName"]:"")."\n\n";
// 2. Define the handler function
$code .= "def $functionName (data, error) :\n";
$code .= " ".implode("\n ",explode("\n",$onUpdate_pyCode))."\n";
// 3. Pass the data
$code .= "data = ".json_encode($unprocessedRow)."\n";
$code .= "error = { \"status\": \"ok\", \"message\": \"\"}\n";
// 4. Call the handler function
$code .= "$functionName(data, error)\n";
// 5. Data changing code
$code .= "\nprint(json.dumps(data))\n";
file_put_contents(__DIR__."/test_code.py",$code);
echo "Running function ".greenText("$functionName(...)")." in sandbox python environment\n";
if (runProcess("/usr/bin/python3",$code,$stdout,$error) != 0) {
sql("UPDATE SYS_PRD_BND.Tables SET LastError = '".str_replace("'",'"',$error)."', LastUpdated = NOW() WHERE Name = '{$activeTable["Name"]}'");
sendTelegramMessage("*ERROR* on trigger function for table {$activeTable["Name"]} : ".$error,"DatabaseGroup");
break;
} else {
// Update database row if needed
$newRowValue = json_decode($stdout,1);
echo "\n".redText("DEBUG new-row-value-json: ").json_encode($newRowValue)."\n";
echo "\n".redText("DEBUG unprocessedRow-json: ").json_encode($unprocessedRow)."\n";
if (json_encode($newRowValue) != json_encode($unprocessedRow)) {
$pkColsName = getTblPrimaryKeyColName($activeTable["Name"]);
$pkColsValues = array_map(fn($cName)=>$unprocessedRow[$cName],$pkColsName);
$sql_instruction = "";
$sql_instruction .= "UPDATE ";
$sql_instruction .= $activeTable["Name"];
$sql_instruction .= " SET ".implode(",",array_map(fn($k,$v)=>"$k=".(is_numeric($v)?$v:"\"$v\""),array_keys($newRowValue),array_values($newRowValue)));
$sql_instruction .= " WHERE ".implode(" AND ",array_map(fn($k,$v)=>"$k = ".(is_numeric($newRowValue[$k])?$newRowValue[$k]:'"'.$newRowValue[$k].'"'),$pkColsName,$pkColsValues));
echo "\n".redText("DEBUG sql-instruction: ").$sql_instruction."\n";
sql($sql_instruction);
}
// Update the status of the operation
sql("UPDATE SYS_PRD_BND.Tables SET LastError = '', LastUpdated = NOW() WHERE Name = '{$activeTable["Name"]}'");
}
}
}
}
}
function greenText($string) { $green = "\033[0;32m"; $reset = "\033[0m"; return $green . $string . $reset ; }
function blueText($string) { $green = "\033[0;34m"; $reset = "\033[0m"; return $green . $string . $reset ; }
function redText($string) { $green = "\033[0;31m"; $reset = "\033[0m"; return $green . $string . $reset ; }
function sendTelegramMessage($message, $dstUsers) {
global $BOT_TOKEN, $CHAT_IDS;
if (is_string($dstUsers)) $dstUsers = [$dstUsers];
foreach($dstUsers as $dstUser) {
if (!isset($CHAT_IDS[$dstUser]))
continue;
else
$CHAT_ID = $CHAT_IDS[$dstUser];
$JSON_RAW_DATA = json_encode([
'chat_id' => $CHAT_ID,
'text' => $message,
'parse_mode' => 'markdown'
]);
$curl = curl_init();
curl_setopt($curl, CURLOPT_URL, "https://api.telegram.org/bot$BOT_TOKEN/sendMessage");
curl_setopt($curl, CURLOPT_HTTPHEADER, array('Content-Type: application/json'));
curl_setopt($curl, CURLOPT_POST, 1);
curl_setopt($curl, CURLOPT_POSTFIELDS, $JSON_RAW_DATA);
curl_setopt($curl, CURLOPT_RETURNTRANSFER, true);
$response = curl_exec($curl);
//print_r($response);
curl_close($curl);
}
}

68
sys.php Executable file
View File

@ -0,0 +1,68 @@
<?php
/*
*
* 3. SYS-Level
*
*
*/
function sql_read($query) {
global $db;
if (!isset($db) || is_null($db)) $db = new PDO("mysql:host=".DB_HOST.";dbname=".DB_NAME,DB_USER,DB_PASS);
$stmt = $db->prepare($query);
if ($stmt->execute() === false) return ["status" => "error", "query"=>$query];
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
function sql_write($query) {
global $db;
if (!isset($db) || is_null($db)) $db = new PDO("mysql:host=".DB_HOST.";dbname=".DB_NAME,DB_USER,DB_PASS);
$stmt = $db->prepare($query);
if ($stmt->execute() === false) die ("sql error: $query");
return $stmt->rowCount();
}
function sql($query) {
if (strpos($query, "SELECT ") === 0) return sql_read($query);
else return sql_write($query);
}
function runProcess($cmd, $stdin, &$stdout, &$stderr) {
$descriptorspec = array(
0 => array("pipe", "r"), // stdin is a pipe that the child will read from
1 => array("pipe", "w"), // stdout is a pipe that the child will write to
2 => array("pipe", "w") // stderr is a pipe that the child will write to
);
$process = proc_open($cmd, $descriptorspec, $pipes);
if (is_resource($process)) {
// Write to stdin and close it
fwrite($pipes[0], $stdin);
fclose($pipes[0]);
// Read the output of the command
$stdout = stream_get_contents($pipes[1]);
fclose($pipes[1]);
// Read the error output of the command
$stderr = stream_get_contents($pipes[2]);
fclose($pipes[2]);
// It's important to close all pipes before calling proc_close in order to avoid a deadlock
$return_value = proc_close($process);
return $return_value;
} else {
// Return an error code if the process could not be started
return -1;
}
}
function getTblPrimaryKeyColName($tblName) {
$dbName = explode(".",$tblName)[0];
$tblName = explode(".",$tblName)[1];
$cols = [];
foreach(sql("SELECT COLUMN_NAME PksColName FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = '$dbName' AND TABLE_NAME = '$tblName' AND CONSTRAINT_NAME = 'PRIMARY';") as $row)
$cols[] = $row["PksColName"];
return $cols;
}