|
use strict;
my $start_time=time;
use Net::FTP;
use DBI;
#use YAML::AppConfig;
use Config::IniFiles;
use threads;
use threads::shared;
use Thread::Semaphore;
use Fcntl qw/:flock/;
my $conf_file='conf.ini';
#my ($ref_host,$ref_username,$ref_passwd,$ref_home,$max_threads)=&get_server_config($conf_file);
my ($ref_host,$ref_username,$ref_passwd,$ref_port,$ref_home)=&get_server_info();
my ($max_threads,$suffix_filter_hash,$fileName_filter_hash) = &get_config($conf_file);
$|=1;
#my @file_array:shared;
my @file_array;
my @result;
#my $id:shared;
my $ftp;
my $ref_files;
my $host;
my @threads;
#运行前清理上次记录,包括文本，xml，数据库等
&clean_up();
#控制线程数
my $running_threads=Thread::Semaphore->new($max_threads);
foreach $ host(@$ref_host)
{
$running_threads->down;
my $thread=threads->create(\&_main,$host,$ref_username,$ref_passwd,$ref_port,$ref_home);#创建线程
push @threads,$thread;#将线程对象压入线程数组
}
while(my $thread=shift @threads)
{
$ref_files=$thread->join(); #problem here $$thread系线程对象
#&show($ref_files); #也是放这里才行
#print "should be two line\n";
push @result,$ref_files if defined $ref_files;
while(my $ref=shift @result)
{
#print "write into file\n";
#&out_to_txt($ref);
&write_db($ref);
}
}
#while(my $ref=shift @result)
# {
# print "write into file\n";
# &out_to_txt($ref);
# &write_db($ref);
# }
print "run time is:",time-$start_time,"seconds\n";
#;
#----------主进程结束---------------------
sub _main
{
my ($host,$ref_username,$ref_passwd,$ref_port,$ref_home)=@_;
print "searching $host\n";
$ftp=&connect_server($host,shift @$ref_username,shift @$ref_passwd,shift @$ref_port);
if(!$ftp)
{
return undef;
}
eval{$ref_files=&get_list(shift @$ref_home,$host);};
if($@)
{
&err($@);
return;
}
#$ref_files=&filter($ref_files,$suffix_filter_hash,$fileName_filter_hash);#过滤处理结果
$ftp->quit;
$running_threads->up;
return $ref_files;
}
#id,name,path,site,suffix
sub get_server_config
{
my $yaml=shift;
my $conf=YAML::AppConfig->new(file=>$yaml);
die "can't find $yaml:$!\n" if !$conf;
my (@host,@username,@passwd,@home,$maxthreads);
$maxthreads=$conf->get('maxthreads')||65;
my $ref_array=$conf->get('server');
foreach my $node(@$ref_array) #遍历所有ftp节点
{
push @host,$node->{'host'};
push @username,$node->{'username'};
push @passwd,$node->{'passwd'};
push @home,$node->{'home'};
}
return (\@host,\@username,\@passwd,\@home,$maxthreads);
}
sub get_config
{
my $cfg_file=shift;
my %suffix_filter_hash;
my %fileName_filter_hash;
my $max_threads;
my $conf=Config::IniFiles->new( -file => $cfg_file);
die "can't find $cfg_file:$!\n" if !$conf;
%suffix_filter_hash=map {$_=>1} split / /,$conf->val('General', 'suffix_filter');
%fileName_filter_hash=map {$_=>1} split / /,$conf->val('General', 'file_name_filter');
$max_threads=$conf->val('General','maxthreads');
return ($max_threads,\%suffix_filter_hash,\%fileName_filter_hash);
}
sub connect_server
{
my($host,$username,$passwd,$port)=@_;
my $try_times=1;
my $ftp;
while(1)
{
#my $ftp=Net::FTP->new($host,Debug=>0,Timeout=>60) or warn "Cannot connect to $host:$@\n";
$ftp=Net::FTP->new($host,Debug=>0,Timeout=>120,Port =>$port,Passive => 0);
if(!$ftp)
{
print "connect server:$host faild,i will try again in 2s later\n";
++$try_times;
#die "Cannot connect to $host:$@" if $try_times>3; #只尝试3次，3次不行，终止线程
if($try_times>3)
{
print "Cannot connect to $host:$@\n";
&err($@);
return undef;
}
sleep 2;
redo;
}
last;
}
$ftp->login($username,$passwd) or &err($ftp->message);
return $ftp;
}
sub get_list
{
my $path=shift;
my @dir=($path);
my $host=shift;
my ($fname,$folder,$site,$suffix);
while($folder=pop(@dir))
{
$ftp->cwd($folder) or die "can't cwd:",$ftp->message;
#print "dir is:$folder\n";
foreach my $line($ftp->dir)
{
next unless my($type,$file_name,$size,$file_time)=&parse_listing($line);
next if $file_name=~/(\.|\.\.)$/;
$file_time=&format_date($file_time);
if($type eq '-')
{
($fname,$suffix)=&parse_suffix($file_name);
if(!&filter($folder,$fname,$suffix,$size,$suffix_filter_hash,$fileName_filter_hash))
{
print "跳过文件:",$host.$folder,$fname,'.',$suffix,"\n";
next;
}
push @file_array,shared_clone([$host,$folder,$fname,$suffix,$size,$file_time]);
}
if($type eq 'd')
{
if($file_name=~/Program Files|新建/)
{
print "跳过目录:",$host,$folder,$file_name,"\n";
next;
}
push @dir,$folder.$file_name.'/';
}
}
#$ftp->cwd($curr) or die "can't cwd:",$ftp->message;
#return \@result;
}
return \@file_array;
}
sub write_db
{
my $ref_file_array=shift;
my $dbh=&db_connect();
#id,site,folder,fileName,suffixId(FK),size,time,hits
my $sql=q{insert into file(site,folder,fileName,suffix,size,fileTime) values(?,?,?,?,?,?)};
my $sth=$dbh->prepare("$sql");
foreach my $row(@$ref_file_array)
{
#next if !defined map $_,@$row;
my ($host,$folder,$fname,$suffix,$size,$file_time)=@$row;
next if !$host&&!$folder&&!$fname&&!$file_time;
#print "@$row,will be writen\n";
#print "site:$host\n";
#$sth->execute($id,$name,$path,$site,$suffix) or die "can't insert into database:",$sth->errstr;
$sth->execute($host,$folder,$fname,$suffix,$size,$file_time); #or die "can't insert into database:",$sth->errstr;
#忽略错误行
if($sth->errstr)
{
&err($sth->errstr);
next;
#print "insert into database faild, data is rollbacking..........\n";
#my $rc=$dbh->rollback()||die $dbh->errstr;
#$dbh->disconnect;
#die $sth->errstr;
}
}
$dbh->commit||die $dbh->errstr;
$dbh->disconnect;
}
sub parse_listing
{
my $list=shift;
return unless my ($type,$mode,$size,$file_time,$name)=
$list=~/^([a-z-])([a-z-]{9}) #权限位，是否为目录
\s+\d* #空格和无聊的数字
(?:\s+\w+){2} #
\s+(\d
+)
\s+(\w+\s+\d+)\s+[\d:]+ #匹配时间
\s(.+) #匹配文件名
$/x;
return ($type,$name,$size,$file_time);
}
sub format_date
{
#Nov 12
my $date_string=shift;
my ($moon,$date)=split /\s+/,$date_string;
my %moon=(
'Jan' => '01',
'Feb' => '02',
'Mar' => '03',
'Apr' => '04',
'May' => '05',
'Jun' => '06',
'Jul' => '07',
'Aug' => '08',
'Sep' => '09',
'Oct' => '10',
'Nov' => '11',
'Dec' => '12',
);
my $year=(localtime)[5]+1900;
#$year+=1900;
$moon=$moon{$moon};
# print $year;
# return $date.' '.$year;
return $year.'-'.$moon.'-'.$date;
}
sub format_size
{
my $size=shift;
return $size;
}
sub parse_suffix
{
my $file_name=shift;
my ($name,$suffix);
$file_name=~/(.*)\.(.*)/gi?return ($1,$2):return ($file_name,undef);
}
sub out_to_txt
{
my $ref_file=shift;
my $line;
open OUT,"+>>",'out.txt' or die "can't open file for write:$!\n"; #这里要注意了，要用“+>>”这个模式
foreach my $row(@$ref_file)
{
#my ($id,$name,$path,$site,$suffix)=map $_,@$row;
#$line=join '>',@$row if defined @$row;
print OUT join '>',@$row if defined @$row ;
print OUT "\n";
#print OUT $line,"\n" ;
}
close OUT or die "can't close file:$!\n";
}
sub filter
{
##数据过滤方案：
#
##数据过滤在介绍数据采集过程时有了初步的介绍，总结一下，我们分几个级别来进行过滤：
#
##第一级别，不区分文件和目录，我们过滤掉所有包含有特定字眼的信息，如黄色、色情等；过滤掉那些长度超过数据库中定义的相关字段
#
## 最大长度的信息。
#
##第二级别，过滤某些目录，比如bin/，Program Files/，新建文件夹/等，其中包含的信息，往往并不是人们需要的。
#
##过滤掉某些文件，其中有一个办法就是通过后缀过滤掉好多文件，另外，也可以通过文件名来过滤，比如readme，新建 Microsoft Word 文档
#
## 新建 文本文档等，他们通常也是一无所用的。
#
##第三级别，前两个级别都是在数据过滤阶段进行，但是这个级别是数据入库之前。这里也是为了过滤掉某些文件，但是这些文件有一些特别：
#
## 只是它们的命名没有任何意义。比如纯数字命名，但是它们的目录往往是非常有用的，所以它们自身被过滤掉，而目录保留下来。
#&filter($folder,$fname,$suffix,$size,$suffix_filter_hash,$fileName_filter_hash);
my ($folder,$fname,$suffix,$size,$suffix_hash,$fileName_hash)=@_;
return 1 if &chech_length($folder,$suffix,$size) and &check_suffix($suffix,$suffix_hash) and &check_fileName($fname,$fileName_hash);
return ;
}
sub chech_length
{
my ($folder,$suffix,$size)=@_;
if(((length $folder == 0 or length $folder>3000)) or ((length $suffix)>10) or ((length $size)>20))
{
#print "跳过文件:",$site.$folder.'/',$name,'.',$suffix,"\n";
return;
}
#print "check length ok\n";
return 1;
}
sub check_suffix
{
my ($suffix,$suffix_hash)=@_;
$suffix=lc $suffix; #if $suffix=~/\w+/;
return if exists $suffix_hash->{$suffix};
#print "check suffix ok\n";
return 1;
}
sub check_fileName
{
my ($fname,$file_name_hash)=@_;
$fname=lc $fname; #if $fname=~/\w+/;
return if exists $file_name_hash->{$fname};
#print "check filename ok\n";
return 1;
}
#sub set_suffix_hash
#{
# my ($ref_hash,$suffix)=@_;
#
# if(defined $suffix)
# {
# chop($suffix);
# if(!exists $suffix_hash->{$suffix})
# {
# $suffix_hash->{$suffix}=length $suffix;
# }
# }
# return $ref_hash;
#}
sub get_server_info
{
my $dbh=&db_connect();
my $sql=q{select site,userName,passWd,port,home from server};
my $sth=$dbh->prepare("$sql");
$sth->execute();
if($sth->errstr)
{
&err($sth->errstr);
$dbh->disconnect;
#return;
die;
}
my (@host,@username,@passwd,@port,@home);
my ($host,$username,$passwd,$port,$home);
while(($host,$username,$passwd,$port,$home)=$sth->fetchrow_array)
{
push @host,$host;
push @username,$username;
push @passwd,$passwd;
push @port,$port;
push @home,$home;
}
$dbh->disconnect;
return (\@host,\@username,\@passwd,\@port,\@home);
}
sub clean_up
{
unlink "out.txt" if -e 'out.txt';
my $dbh=&db_connect();
my $truncate_table_sql=q{truncate table `ftpsearch`.`file`};
my $auto_increat_sql=q{alter table file AUTO_INCREMENT=1};
my $sth=$dbh->prepare("$truncate_table_sql");
$sth->execute();
if($sth->errstr)
{
&err($sth->errstr);
my $rc=$dbh->rollback()||die $dbh->errstr;
$dbh->disconnect;
die;
}
$dbh->do($auto_increat_sql);
$dbh->commit||die $dbh->errstr;
$dbh->disconnect;
}
sub db_connect
{
my $database='DBI:mysql:database=ftpsearch;host=127.0.0.1';
my $user='ftpCollect';
my $pw='passwd';
my $dbh=DBI->connect($database,$user,$pw,{'RaiseError'=>0,AutoCommit=>0});
$DBI::errstr?(&err($DBI::errstr),die):return $dbh;
}
sub err
{
chomp(my $msg=shift);
#my $sucdess_log_file='.\\log\\SC_log.log';
#my $err_file='D:\\test-area\\search\\log\\error.log';
my $err_file='log\\error.log';
open ERR,">>",$err_file;
flock(ERR,LOCK_EX)||die "can't get lock:$!\n";
#my $date=localtime;
#print ERR '[',$date,']',$msg,"\n";
print ERR '[',scalar(localtime),']',$msg,"\n";
flock(ERR,LOCK_UN)||die "can't unlock:$!\n";
close ERR;
}
#——BY Weigun http://wfnh.cublog.cn/
|
