create or replace and compile java source named pump_csv as
import java.sql.*;
import java.text.*;
import java.lang.*;
import java.util.*;
import java.io.*;
import oracle.jdbc.*; // Подгружаем драйвер ORACLE
import c_pump_tools.*;
public class c_pump_csv
{
static boolean DEBUG = true; // Включает режим пассивной отладки (вывод диагностических сообщений)
static String CSV_DIR_NAME = "CSV_DIR";
static String ORA_URL = "jdbc:default:connection"; // URL соединения с MSSQL
public static void run() throws Exception
{
// Грузим драйвер ORACLE
DriverManager.registerDriver(new oracle.jdbc.OracleDriver());
OracleConnection conORA = null;
try
{
// Соединяемся с базой данных ORACLE по внутреннему соединению
try
{
conORA = (OracleConnection)DriverManager.getConnection(ORA_URL);
}
catch(Exception e)
{
e.printStackTrace();
throw new Exception("Невозможно соединиться с сервером-приемником данных по URL=" + ORA_URL);
}
conORA.setAutoCommit(false);// Устанавливаем отказ от autocommit
log("Connected ORACLE on " + ORA_URL);
String csvDirectoryName = null;
OraclePreparedStatement stmtORA_InsertData = null;
OracleStatement stmtORA_CheckStageClearance = null;
try
{
// Получаем имя директории с CSV файлами
try
{
OracleStatement stmtORA_GetDirectory = null;
OracleResultSet ors=null;
try
{
stmtORA_GetDirectory = (OracleStatement)conORA.createStatement();
ors = (OracleResultSet)stmtORA_GetDirectory.executeQuery("select directory_path from dba_directories where directory_name = '"+CSV_DIR_NAME+"'");
if(ors.next()) csvDirectoryName = ors.getString(1);
if (csvDirectoryName == null)
{
log("Unable to resolve directory name " + CSV_DIR_NAME + ". Pumping cannot be initiated");
throw new Exception ("Невозможно разрешить имя директории '" + CSV_DIR_NAME + "'");
}
log("Path to \"" + csvDirectoryName + "\" has been found to pump from");
}
finally
{
if (ors instanceof OracleResultSet)ors.close();
if (stmtORA_GetDirectory instanceof OracleStatement) stmtORA_GetDirectory.close();
log("Resources for resolving the CSV directory name has been released");
}
}
catch(Exception e)
{
e.printStackTrace();
throw new Exception("Невозможно получить путь к папке по имени директории " + CSV_DIR_NAME);
}
// Проверяем готовность целевой таблицы к приему новых данных
try
{
stmtORA_CheckStageClearance = (OracleStatement)conORA.createStatement();
OracleResultSet rstCheck = null;
try
{
rstCheck = (OracleResultSet)stmtORA_CheckStageClearance.executeQuery("select count(*) cnt from target_table");
int iCountRows = 0;
if (rstCheck.next()) iCountRows = rstCheck.getInt(1);
if (iCountRows>0)
{
log("WARNING! Source table TARGET_TABLE is not empty. Data cannot be loaded into non-empty table. Terminating data import");
return;
}
log("Source table TARGET_TABLE is empty and ready to receive data");
}
finally
{
if (rstCheck != null) rstCheck.close();
}
}
catch(Exception e)
{
e.printStackTrace();
throw new Exception("Невозможно выполнить команду для проверки готовности целевой таблицы TARGET_TABLE сервера-приемника");
}
// Просматриваем содержимое папки на предмет наличия в ней CSV файлов
File sourceFolder = new File(csvDirectoryName);
FileFilter sourceFolderFilter = new FileFilter()
{
public boolean accept(File file)
{
return !file.isDirectory()&& file.getName().endsWith(".csv");
}
};
File[]csvFiles = sourceFolder.listFiles(sourceFolderFilter);
if (csvFiles == null || csvFiles.length == 0)
{
log("Source CSV files not found in directory " + CSV_DIR_NAME + ". Exiting process");
return;
}
else
{
for (int iterFile=0; iterFile
{
// Открываем и построчно читаем текстовый файл
File csvFile = csvFiles[iterFile];
log("File " + csvFile.getCanonicalPath() + " has been found in source SSV folder");
BufferedReader input = new BufferedReader(new FileReader(csvFile));
try
{
// Инициализируем переменные для разбора строк
StringBuffer rowColumnNames = new StringBuffer();
StringBuffer rowColumnBinds = new StringBuffer();
int columnCount = 0;
String line = null;
long rowIndex = 0;
try
{
while (( line = input.readLine()) != null)
{
rowIndex++;
if(rowIndex == 1)
{
// В первой строке находятся столбцы, для вставки данных
StringTokenizer st = new StringTokenizer(line, ";");
while(st.hasMoreTokens())
{
String fieldName = st.nextToken().trim().toUpperCase();
if (rowColumnNames.length() > 0) rowColumnNames.append(",");
rowColumnNames.append(fieldName);
if (rowColumnBinds.length() > 0) rowColumnBinds.append(",");
rowColumnBinds.append("?");
columnCount++;
}
if (columnCount==0) throw new Exception("В первой строке CSV файла не обнаружены названия столбцов. Импорт данных невозможен.");
log("Found " + new Integer(columnCount).toString() +" columns: " + rowColumnNames.toString() );
// Готовим команду для вставки данных
try
{
stmtORA_InsertData = (OraclePreparedStatement)conORA.prepareStatement("insert into target_table("+rowColumnNames.toString()+") values("+rowColumnBinds.toString()+")");
}
catch(Exception e)
{
e.printStackTrace();
throw new Exception("Невозможно подготовить команду для вставки данных из CSV источника");
}
}
else
{
// Разбор данных
StringTokenizer st = new StringTokenizer(line, ";");
ArrayList columnValues = new ArrayList(columnCount);
while(st.hasMoreTokens())
{
columnValues.add(st.nextToken());
}
if (columnValues.size() != columnCount) throw new Exception("Количество значений (" + columnValues.size()+ ") не соответствует количеству столбцов ("+new Integer(columnCount).toString()+")");
// Вставка данных в БД
if (columnCount>=1)
{
try{
stmtORA_InsertData.setString(1, (String)columnValues.get(0));
}catch(Exception e) {log("Error parsing data cell #1");throw e;}
}
if (columnCount>=2)
{
try{
stmtORA_InsertData.setString(2, (String)columnValues.get(1));
}catch(Exception e) {log("Error parsing data cell #2");throw e;}
}
if (columnCount>=3)
{
try{
stmtORA_InsertData.setString(3, (String)columnValues.get(2));
}catch(Exception e) {log("Error parsing data cell #3");throw e;}
}
if (columnCount>=4)
{
try{
stmtORA_InsertData.setString(4, (String)columnValues.get(3));
}catch(Exception e) {log("Error parsing data cell #4");throw e;}
}
if (columnCount>=5)
{
try{
stmtORA_InsertData.setString(5, (String)columnValues.get(4));
}catch(Exception e) {log("Error parsing data cell #5");throw e;}
}
if (columnCount>=6)
{
try{
stmtORA_InsertData.setString(6, (String)columnValues.get(5));
}catch(Exception e) {log("Error parsing data cell #6");throw e;}
}
if (columnCount>=7)
{
try{
stmtORA_InsertData.setString(7, (String)columnValues.get(6));
}catch(Exception e) {log("Error parsing data cell #7");throw e;}
}
if (columnCount>=8)
{
try{
stmtORA_InsertData.setString(8, (String)columnValues.get(7));
}catch(Exception e) {log("Error parsing data cell #8");throw e;}
}
stmtORA_InsertData.execute();
}
} //while (read file)
stmtORA_InsertData.execute("commit");
log("Processed " + new Long(rowIndex).toString() + " rows. Pumping finished");
}
catch(Exception e)
{
e.printStackTrace();
throw new Exception("\n\nФайл "+csvFile.getName()+"\nСбой закачки в строке " + rowIndex + ":\n"+e.getMessage()+"\n\n");
}
}
finally
{
if (stmtORA_InsertData instanceof OraclePreparedStatement) stmtORA_InsertData.close();
input.close();
log("Input CSV stream closed");
csvFile.delete();
log("Source CSV file deleted");
}
} //iterate csvFiles
} // csvFiles is not null
}
finally
{
// Освобождаем все используемые ресурсы
if (stmtORA_InsertData != null) stmtORA_InsertData.close();
if (stmtORA_CheckStageClearance != null) stmtORA_CheckStageClearance.close();
log("Resources released");
}
}
catch(Exception e)
{
log(e.getMessage());
throw e;
}
finally
{
// Закрываем соединения с базами данных
if (conORA != null && !conORA.isClosed()) conORA.close();
log(Connections closed");
}
}
// Пишем в LOG сервера
public static void log(String msg)
{
if (DEBUG) c_pump_tools.log(msg);
}
}
